scripts/discover_walkthrough.py

#!/usr/bin/env python3
"""Drive Discover the way a real user would, flagging weird kinks card-by-card.

Loops:
  1. Fetch the next recommendation card from ``GET /users/{id}/recommendations``.
  2. Inspect the card (name, content_kind, definition, summary, popularity, etc.).
  3. Pick a reaction with a deterministic strategy AND record any anomalies that a real user
     would notice (long prose names, all-lowercase short names, empty descriptions, suspected
     duplicates of an earlier card, geo/city noise, scenario-shaped titles, etc.).
  4. Save the play via ``POST /users/{id}/plays``.
  5. Repeat ``--turns`` times.

End-of-run, write a JSON report and a terse summary so we can fix any patterns we see.

Examples:
    python scripts/discover_walkthrough.py --base-url https://perplexed7675-kink-discovery.hf.space --turns 120
    python scripts/discover_walkthrough.py --base-url http://127.0.0.1:8012 --turns 60 --strategy automated
"""
from __future__ import annotations

import argparse
import json
import re
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from collections import Counter, defaultdict
from dataclasses import dataclass, field
from pathlib import Path

DEFAULT_BASE_URL = "http://127.0.0.1:8012"
HTTP_TIMEOUT_S = 60


@dataclass
class Card:
    kink_id: str
    name: str
    content_kind: str
    is_scenario: bool
    summary: str
    detail_summary: str
    definition: str
    notes: str
    popularity: float
    cluster: str
    discovery_source: str
    reasons: list[str]
    asset_count: int
    raw: dict[str, object]


@dataclass
class Anomaly:
    code: str
    detail: str


@dataclass
class Turn:
    n: int
    card: Card
    rating: str
    directions: list[str]
    anomalies: list[Anomaly] = field(default_factory=list)


def _http(method: str, url: str, *, headers: dict[str, str] | None = None, body: bytes | None = None) -> tuple[int, dict[str, object]]:
    request = urllib.request.Request(url, data=body, method=method, headers=headers or {})
    try:
        with urllib.request.urlopen(request, timeout=HTTP_TIMEOUT_S) as response:
            raw = response.read().decode("utf-8")
            payload = json.loads(raw) if raw else {}
            return response.status, payload
    except urllib.error.HTTPError as exc:
        text = exc.read().decode("utf-8", errors="ignore") if exc.fp else ""
        try:
            payload = json.loads(text) if text else {"error": str(exc)}
        except json.JSONDecodeError:
            payload = {"error": text}
        return exc.code, payload


def _post_json(base: str, path: str, body: dict, headers: dict[str, str]) -> tuple[int, dict[str, object]]:
    return _http("POST", base + path, headers={**headers, "Content-Type": "application/json"}, body=json.dumps(body).encode("utf-8"))


def _get_json(base: str, path: str, headers: dict[str, str] | None = None) -> tuple[int, dict[str, object]]:
    return _http("GET", base + path, headers=headers)


def create_user(base: str) -> tuple[str, str]:
    code, payload = _post_json(base, "/users", {}, headers={})
    if code != 200:
        raise RuntimeError(f"create_user failed: {code} {payload}")
    return str(payload["id"]), str(payload["private_token"])


def get_user(base: str, uid: str, token: str) -> dict[str, object]:
    code, payload = _get_json(base, f"/users/{urllib.parse.quote(uid)}", headers={"x-private-token": token})
    if code != 200:
        raise RuntimeError(f"get_user failed: {code} {payload}")
    return payload


def get_recommendations(base: str, uid: str, token: str, limit: int = 24) -> list[dict[str, object]]:
    code, payload = _get_json(base, f"/users/{urllib.parse.quote(uid)}/recommendations?limit={limit}", headers={"x-private-token": token})
    if code != 200:
        raise RuntimeError(f"recommendations failed: {code} {payload}")
    return list(payload.get("items") or [])


def save_play(base: str, uid: str, token: str, kink_id: str, rating: str, directions: list[str]) -> int:
    code, _ = _post_json(
        base,
        f"/users/{urllib.parse.quote(uid)}/plays",
        {"kink_id": kink_id, "interest_state": rating, "directions": directions},
        headers={"x-private-token": token},
    )
    return code


_GEO_TOKENS = {
    "abu", "amsterdam", "atlanta", "austin", "bangkok", "barcelona", "berlin", "boston", "brooklyn",
    "buenos", "cairo", "calgary", "charlotte", "chicago", "cincinnati", "cleveland", "columbus",
    "copenhagen", "dallas", "denver", "detroit", "dubai", "dublin", "edinburgh", "edmonton",
    "frankfurt", "geneva", "glasgow", "helsinki", "hong", "houston", "indianapolis", "istanbul",
    "jacksonville", "johannesburg", "kansas", "kolkata", "kuala", "vegas", "lisbon", "london",
    "louisville", "madrid", "manchester", "manhattan", "melbourne", "memphis", "miami", "milan",
    "milwaukee", "minneapolis", "montreal", "moscow", "mumbai", "nashville", "delhi", "orleans",
    "york", "oakland", "omaha", "orlando", "oslo", "ottawa", "paris", "philadelphia", "pittsburgh",
    "portland", "prague", "raleigh", "sacramento", "antonio", "diego", "francisco", "jose", "sao",
    "seattle", "shanghai", "singapore", "stockholm", "sydney", "tampa", "tokyo", "toronto", "tulsa",
    "vancouver", "vienna", "warsaw", "winnipeg", "zurich",
}

_PROFILE_SENTENCE_STARTS = ("i love ", "i like ", "i want ", "i would ", "i enjoy ")
_BAD_TYPOGRAPHY_RE = re.compile(r"_(?=[A-Za-z])|(?<=[A-Za-z])_")


def _norm_compact(s: str) -> str:
    return re.sub(r"[^a-z0-9]", "", s.lower())


def detect_anomalies(card: Card, prior_compacts: dict[str, str]) -> list[Anomaly]:
    out: list[Anomaly] = []
    name = card.name or ""
    lower = name.lower().strip()

    if not name.strip():
        out.append(Anomaly("empty_name", "card has no name"))
        return out

    if card.is_scenario:
        out.append(Anomaly("scenario_in_deck", "is_scenario=True surfaced in Discover"))

    if card.content_kind and card.content_kind != "play":
        out.append(Anomaly("non_play_content_kind", f"content_kind={card.content_kind!r}"))

    if len(name) > 40:
        out.append(Anomaly("long_name", f"name length {len(name)} chars — likely scenario or prose"))

    word_count = len(re.findall(r"[A-Za-z0-9]+", name))
    if len(name) >= 35 and word_count >= 5:
        out.append(Anomaly("prose_shape", f"{word_count} words / {len(name)} chars — reads like a sentence"))

    if any(lower.startswith(p) for p in _PROFILE_SENTENCE_STARTS):
        out.append(Anomaly("profile_sentence", f"name starts with {lower.split()[0]!r} — profile prose"))

    # Casing oddities (excluding acronyms which are intentionally upper).
    alpha_only = re.sub(r"[^A-Za-z]", "", name)
    if alpha_only and len(alpha_only) >= 6:
        if alpha_only == alpha_only.lower():
            out.append(Anomaly("all_lowercase", "no caps anywhere — looks like raw user-typed entry"))
        elif alpha_only == alpha_only.upper() and word_count >= 2:
            out.append(Anomaly("all_uppercase", "all-caps multi-word name"))

    if _BAD_TYPOGRAPHY_RE.search(name):
        out.append(Anomaly("underscored_name", "underscores between letters (profile-export shape)"))

    # Geo/city noise
    name_tokens = set(re.findall(r"[a-z]+", lower))
    if name_tokens & _GEO_TOKENS:
        out.append(Anomaly("geo_token", "name contains a city/region token"))

    # Empty body — both summary and detail_summary blank → no help text shown to user.
    if not card.summary.strip() and not card.detail_summary.strip() and not card.definition.strip():
        out.append(Anomaly("no_body_text", "no summary / detail_summary / definition for this card"))

    # Definition shows up in expanded mode; short summary in collapsed. If they're identical,
    # tap-for-full-info gains nothing.
    if card.definition.strip() and card.summary.strip() and card.definition.strip() == card.summary.strip():
        out.append(Anomaly("expand_no_op", "definition equals summary — Tap for full info reveals nothing more"))

    # Suspected duplicate of an already-seen card.
    cf = _norm_compact(name)
    if len(cf) >= 8:
        if cf in prior_compacts and prior_compacts[cf] != card.kink_id:
            out.append(Anomaly("duplicate_of_prior", f"compact_form matches prior card {prior_compacts[cf]!r}"))
        else:
            prior_compacts[cf] = card.kink_id

    return out


def render_card_summary(card: Card) -> str:
    bits = [
        f"#{card.kink_id:<22}  {card.name}",
        f"  popularity={card.popularity:.0f}  asset_count={card.asset_count}  source={card.discovery_source}",
    ]
    if card.definition.strip():
        bits.append(f"  def: {card.definition.strip()[:200]}")
    elif card.summary.strip():
        bits.append(f"  sum: {card.summary.strip()[:200]}")
    else:
        bits.append("  (no description)")
    if card.reasons:
        bits.append(f"  why: {' / '.join(card.reasons[:2])}")
    return "\n".join(bits)


def card_from_item(item: dict[str, object]) -> Card:
    k = item.get("kink") or {}
    return Card(
        kink_id=str(k.get("id", "")),
        name=str(k.get("name", "") or ""),
        content_kind=str(k.get("content_kind", "") or ""),
        is_scenario=bool(k.get("is_scenario")),
        summary=str(k.get("summary", "") or ""),
        detail_summary=str(k.get("detail_summary", "") or ""),
        definition=str(k.get("definition", "") or ""),
        notes=str(k.get("notes", "") or ""),
        popularity=float(k.get("popularity", 0.0) or 0.0),
        cluster=str(k.get("cluster", "") or ""),
        discovery_source=str(item.get("discovery_source", "") or ""),
        reasons=list(item.get("reasons") or []),
        asset_count=len(k.get("assets") or []),
        raw=dict(item),
    )


def automated_choice(card: Card, anomalies: list[Anomaly]) -> tuple[str, list[str]]:
    """Deterministic 'reasonable user' policy that exercises the full reaction set."""
    codes = {a.code for a in anomalies}
    if codes & {"prose_shape", "profile_sentence", "long_name", "scenario_in_deck", "non_play_content_kind", "geo_token", "underscored_name", "all_uppercase", "duplicate_of_prior"}:
        return "hard_no", []
    if codes & {"all_lowercase", "no_body_text", "expand_no_op"}:
        return "not_interested", []
    if card.popularity < 50 and not card.definition.strip():
        return "curious", ["together"]
    if card.popularity >= 5000 and card.definition.strip():
        return "love", ["together"]
    return "like", ["together"]


def manual_choice(card: Card, anomalies: list[Anomaly]) -> tuple[str, list[str]]:
    print(render_card_summary(card))
    if anomalies:
        print("  anomalies:", ", ".join(f"{a.code}({a.detail})" for a in anomalies))
    choice = ""
    while choice not in {"love", "like", "curious", "skip", "ni", "hard_no"}:
        choice = input("  rate [love/like/curious/skip/ni/hard_no]: ").strip().lower()
    return ("not_interested" if choice == "ni" else "skip" if choice == "skip" else choice), ["together"]


def run_walkthrough(base_url: str, turns: int, strategy: str, out_path: Path | None) -> dict[str, object]:
    uid, token = create_user(base_url)
    print(f"[walkthrough] new user: {uid}")
    record_turns: list[Turn] = []
    prior_compacts: dict[str, str] = {}
    seen_kink_ids: set[str] = set()
    anomaly_counter: Counter[str] = Counter()
    consecutive_empty = 0
    turn_idx = 0
    while turn_idx < turns:
        recs = get_recommendations(base_url, uid, token, limit=24)
        items = [item for item in recs if str(((item.get("kink") or {}).get("id") or "")) not in seen_kink_ids]
        if not items:
            consecutive_empty += 1
            if consecutive_empty > 3:
                print(f"[walkthrough] recommendations exhausted at turn {turn_idx}")
                break
            time.sleep(1.0)
            continue
        consecutive_empty = 0
        item = items[0]
        card = card_from_item(item)
        if not card.kink_id:
            time.sleep(0.5)
            continue
        anomalies = detect_anomalies(card, prior_compacts)
        for a in anomalies:
            anomaly_counter[a.code] += 1
        if strategy == "manual":
            rating, dirs = manual_choice(card, anomalies)
            if rating == "skip":
                seen_kink_ids.add(card.kink_id)
                turn_idx += 1
                continue
        else:
            rating, dirs = automated_choice(card, anomalies)
        status = save_play(base_url, uid, token, card.kink_id, rating, dirs)
        if status != 200:
            print(f"  save failed: {status}")
            seen_kink_ids.add(card.kink_id)
            turn_idx += 1
            continue
        seen_kink_ids.add(card.kink_id)
        record_turns.append(Turn(n=turn_idx + 1, card=card, rating=rating, directions=dirs, anomalies=anomalies))
        turn_idx += 1
        if turn_idx % 10 == 0:
            print(f"[walkthrough] turn {turn_idx} / {turns}: anomalies so far -> {dict(anomaly_counter)}")

    summary = {
        "user_id": uid,
        "base_url": base_url,
        "turns_attempted": turns,
        "turns_completed": len(record_turns),
        "anomaly_counts": dict(anomaly_counter),
        "rating_distribution": dict(Counter(t.rating for t in record_turns)),
        "discovery_source_distribution": dict(Counter(t.card.discovery_source for t in record_turns)),
        "content_kind_distribution": dict(Counter(t.card.content_kind for t in record_turns)),
    }
    flagged = []
    for t in record_turns:
        if t.anomalies:
            flagged.append({
                "n": t.n,
                "kink_id": t.card.kink_id,
                "name": t.card.name,
                "rating": t.rating,
                "anomalies": [{"code": a.code, "detail": a.detail} for a in t.anomalies],
                "popularity": t.card.popularity,
                "discovery_source": t.card.discovery_source,
                "reasons": t.card.reasons,
            })
    summary["flagged_turns"] = flagged
    if out_path is not None:
        out_path.write_text(json.dumps(summary, indent=2, sort_keys=True), encoding="utf-8")
    return summary


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
    ap.add_argument("--base-url", default=DEFAULT_BASE_URL)
    ap.add_argument("--turns", type=int, default=120)
    ap.add_argument("--strategy", choices=["automated", "manual"], default="automated")
    ap.add_argument("--out", type=Path, default=Path("/tmp/discover_walkthrough.json"))
    args = ap.parse_args()
    summary = run_walkthrough(args.base_url.rstrip("/"), args.turns, args.strategy, args.out)
    print()
    print("=== walkthrough summary ===")
    print(f"user_id={summary['user_id']}  turns={summary['turns_completed']}/{summary['turns_attempted']}")
    print(f"ratings: {summary['rating_distribution']}")
    print(f"discovery_source: {summary['discovery_source_distribution']}")
    print(f"content_kind: {summary['content_kind_distribution']}")
    print(f"anomalies: {summary['anomaly_counts']}")
    print(f"flagged turns: {len(summary['flagged_turns'])}")
    print(f"detail written to {args.out}")
    return 0


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