import json import os import random import re from functools import lru_cache from pathlib import Path from typing import Any try: from fastapi import Request from fastapi.responses import HTMLResponse, JSONResponse from fastapi.staticfiles import StaticFiles except Exception: # pragma: no cover - app dependencies are installed on Spaces. Request = Any HTMLResponse = str JSONResponse = dict StaticFiles = None try: from gradio import Server as GradioServer except Exception: # pragma: no cover - keeps helper functions importable locally. GradioServer = None try: from fastapi import FastAPI except Exception: # pragma: no cover FastAPI = None BASE_DIR = Path(__file__).resolve().parent DATA_PATH = BASE_DIR / "data" / "taboo_cards_playable.json" SECONDS_PER_ROUND = 30 MAX_CLUE_CHARS = 280 MAX_MODEL_GUESSES = 3 MODEL_ID = "onnx-community/tiny-aya-global-ONNX" if GradioServer is not None: app = GradioServer() elif FastAPI is not None: app = FastAPI(title="Clue Vibes") else: # pragma: no cover raise RuntimeError("Gradio or FastAPI is required to run Clue Vibes.") if StaticFiles is not None: app.mount("/static", StaticFiles(directory=BASE_DIR / "static"), name="static") def _api_endpoint(name: str): if hasattr(app, "api"): return app.api(name=name) def decorator(func): return func return decorator def _plain_text(value: Any) -> str: return re.sub(r"\s+", " ", str(value or "").strip()) def _word_key(value: Any) -> str: return _plain_text(value).lower() def _compact(value: str) -> str: return re.sub(r"[^a-z0-9]+", "", value.lower()) def _variants(word: str) -> set[str]: base = _word_key(word) if not base: return set() variants = {base} if re.fullmatch(r"[a-z]+", base): variants.add(f"{base}s") variants.add(f"{base}es") variants.add(f"{base}ed") variants.add(f"{base}ing") if base.endswith("e"): variants.add(f"{base[:-1]}ed") variants.add(f"{base[:-1]}ing") if base.endswith("y") and len(base) > 2: variants.add(f"{base[:-1]}ies") if len(base) > 2 and base[-1] not in "aeiouy": variants.add(f"{base}{base[-1]}ed") variants.add(f"{base}{base[-1]}ing") return variants def _contains_forbidden(clue: str, forbidden: list[str]) -> dict[str, Any]: clue_key = clue.lower() clue_compact = _compact(clue) for word in forbidden: clean = _word_key(word) if not clean: continue for variant in _variants(clean): if re.search(rf"(?= 4 and compact_word in clue_compact: return {"ok": False, "word": clean} return {"ok": True, "word": ""} def _is_playable_card(card: dict[str, Any]) -> bool: target = _word_key(card.get("target")) banned = card.get("banned") if not re.fullmatch(r"[a-z][a-z-]{2,23}", target): return False if not isinstance(banned, list) or len(banned) < 5: return False return all(re.fullmatch(r"[a-z][a-z-]{2,23}", _word_key(item)) for item in banned[:5]) @lru_cache(maxsize=1) def _cards() -> list[dict[str, Any]]: if not DATA_PATH.exists(): raise FileNotFoundError(f"Card dataset not found at {DATA_PATH}") with DATA_PATH.open("r", encoding="utf-8") as handle: data = json.load(handle) raw_cards = data.get("cards", []) cards = [ { "target": _word_key(card.get("target")), "banned": [_word_key(item) for item in card.get("banned", [])[:5]], } for card in raw_cards if isinstance(card, dict) and _is_playable_card(card) ] if not cards: raise RuntimeError("No playable cards were found in the dataset.") return cards def new_round() -> dict[str, Any]: cards = _cards() index = random.randrange(len(cards)) card = cards[index] return { "ok": True, "card_id": index, "target": card["target"], "banned": card["banned"], "seconds": SECONDS_PER_ROUND, "max_chars": MAX_CLUE_CHARS, "max_guesses": MAX_MODEL_GUESSES, "deck_size": len(cards), "model_id": MODEL_ID, } def validate_clue(target: str, banned: list[str], clue: str) -> dict[str, Any]: target = _word_key(target) banned = [_word_key(item) for item in banned] clue = _plain_text(clue) if not clue: return {"ok": False, "error": "Write a clue first.", "blocked_word": ""} if len(clue) > MAX_CLUE_CHARS: return { "ok": False, "error": f"Keep the clue under {MAX_CLUE_CHARS} characters.", "blocked_word": "", } blocked = _contains_forbidden(clue, [target, *banned]) if not blocked["ok"]: return { "ok": False, "error": "That clue uses a secret or banned word.", "blocked_word": blocked["word"], } return {"ok": True, "error": "", "blocked_word": ""} def score_guess(target: str, guess: str) -> dict[str, Any]: target = _word_key(target) guess_key = _word_key(guess) guess_tokens = set(re.findall(r"[a-z][a-z-]*", guess_key)) target_variants = _variants(target) correct = guess_key in target_variants or bool(guess_tokens & target_variants) return {"ok": True, "correct": correct, "target": target, "guess": guess_key} @_api_endpoint(name="new_round") def gradio_new_round() -> dict[str, Any]: """Start a new Clue Vibes round.""" return new_round() @_api_endpoint(name="validate_clue") def gradio_validate_clue(target: str, banned: list[str], clue: str) -> dict[str, Any]: """Validate that a clue does not include the target or banned words.""" return validate_clue(target, banned, clue) @_api_endpoint(name="score_guess") def gradio_score_guess(target: str, guess: str) -> dict[str, Any]: """Score Tiny Aya's guessed word against the current target.""" return score_guess(target, guess) @app.post("/round") async def round_route() -> JSONResponse: return JSONResponse(new_round()) @app.post("/validate") async def validate_route(request: Request) -> JSONResponse: payload = await request.json() result = validate_clue( target=str(payload.get("target", "")), banned=payload.get("banned") or [], clue=str(payload.get("clue", "")), ) return JSONResponse(result, status_code=200 if result.get("ok") else 422) @app.post("/score_guess") async def score_guess_route(request: Request) -> JSONResponse: payload = await request.json() return JSONResponse(score_guess(str(payload.get("target", "")), str(payload.get("guess", "")))) @app.get("/", response_class=HTMLResponse) async def homepage() -> str: return (BASE_DIR / "index.html").read_text(encoding="utf-8") if __name__ == "__main__": port = int(os.environ.get("PORT", "7860")) if hasattr(app, "launch"): app.launch(server_name="0.0.0.0", server_port=port, show_error=True) else: import uvicorn uvicorn.run(app, host="0.0.0.0", port=port)