""" AI Puzzle Maker — the Gradio app (Hugging Face Space entrypoint). A gr.Blocks page whose entire UI is the custom HTML5 canvas jigsaw game, with same-origin backend routes mounted onto Gradio's FastAPI server: POST /quip -> MiniCPM commentary (live hints, roasts, victory lines) POST /generate -> puzzle generation (MiniCPM theme pack + in-Space FLUX.2-klein art) POST /publish -> share a puzzle to the community gallery GET /puzzles -> gallery listing GET /puzzle/{id} -> full pack POST /like -> like a puzzle GET /times -> per-puzzle solve-time leaderboard POST /times -> submit a solve Run locally: pip install -r requirements.txt gradio && python app.py -> http://localhost:7860 Deploy: push to a Hugging Face Space (SDK: gradio, app_file: app.py) """ import json import os import re import urllib.request import gradio as gr from fastapi import FastAPI, Request from fastapi.staticfiles import StaticFiles from starlette.concurrency import run_in_threadpool import flux_local # FLUX.2-klein-9B on ZeroGPU (AVAILABLE=False on machines without GPU/torch) import community # opt-in shared puzzles on the mounted bucket import leaderboard # per-puzzle solve times on the mounted bucket import moderation # Nemotron-3.5-Content-Safety via a Modal GPU endpoint (fail-open if unavailable) import prompt_pipeline # MiniCPM writes the per-subject puzzle pack (title, voice, prompts, quips) HERE = os.path.dirname(os.path.abspath(__file__)) STATIC = os.path.join(HERE, "static") # ---------------- MiniCPM (free hackathon commentary API) ---------------- MINICPM_URL = os.environ.get("MINICPM_URL", "") # secrets — set via env / Space secrets, never in source MINICPM_KEY = os.environ.get("MINICPM_KEY", "") MINICPM_MODEL = "MiniCPM4.1-8B" _THINK = re.compile(r"[\s\S]*?", re.IGNORECASE) def minicpm(messages, max_tokens=80, temperature=1.0): if not (MINICPM_URL and MINICPM_KEY): raise RuntimeError("MINICPM_URL/MINICPM_KEY not set — quips fall back to canned lines") body = json.dumps({ "model": MINICPM_MODEL, "messages": messages, "max_tokens": max_tokens, "temperature": temperature, "chat_template_kwargs": {"enable_thinking": False}, # MiniCPM4.1 is a thinking model }).encode() req = urllib.request.Request( MINICPM_URL, data=body, headers={"Authorization": "Bearer " + MINICPM_KEY, "Content-Type": "application/json"}, ) with urllib.request.urlopen(req, timeout=20) as r: data = json.loads(r.read()) return _THINK.sub("", data["choices"][0]["message"]["content"]).strip() def generate_pack(subject): """Subject -> MiniCPM puzzle pack -> in-Space Flux artwork + mascot. Returns: { title, desc, voice, mascot_name, quips, victory, subject, image (jpeg data-URL), thumb (jpeg data-URL), mascot (png data-URL) } `title`/`voice`/`quips` are always returned (themed commentary works everywhere); `image`/`mascot` are filled only where FLUX.2-klein is live (the ZeroGPU Space) — the client falls back to a bundled puzzle when `image` is empty. """ subject = (subject or "a cozy log cabin in snowy mountains at dusk").strip() safe, cats = moderation.check(subject) # screen BEFORE generating if not safe: return {"error": f"blocked by content safety ({cats or 'policy violation'}) — try a different idea"} try: pack = prompt_pipeline.write_puzzle_pack(subject) except Exception as e: print(f"theme pack failed ({e}); using enforced defaults") pack = prompt_pipeline._enforce({}, subject) out = {"title": pack["title"], "desc": pack["desc"], "voice": pack["voice"], "mascot_name": pack["mascot_name"], "quips": pack["quips"], "victory": pack["victory"], "subject": subject, "image": "", "thumb": "", "mascot": ""} if flux_local.AVAILABLE: # in-Space FLUX.2-klein-9B on ZeroGPU import io import sprites imgs = flux_local.gen_images([(pack["prompts"]["image"], 1024, 768), (pack["prompts"]["mascot"], 512, 512)]) art = sprites.process_artwork(_png_bytes(imgs[0], io)) out["image"] = sprites.to_jpeg_data_url(art) out["thumb"] = sprites.to_jpeg_data_url(sprites.thumb(art), quality=80) try: out["mascot"] = sprites.to_data_url(sprites.process_mascot(_png_bytes(imgs[1], io))) except Exception as e: print(f"mascot processing failed (artwork still fine): {e}") return out def _png_bytes(img, io): buf = io.BytesIO() img.save(buf, "PNG") return buf.getvalue() # ---------------- backend routes (same-origin for the embedded game) ---------------- api = FastAPI() @api.post("/quip") async def quip(req: Request): p = await req.json() try: mt = max(1, min(int(p.get("max_tokens", 80)), 220)) # clamp client input — shared API tp = max(0.0, min(float(p.get("temperature", 1.0)), 1.5)) # threadpool: the urllib call is blocking; never stall the whole server on it text = await run_in_threadpool(minicpm, p.get("messages", []), mt, tp) return {"text": text} except Exception as e: return {"text": "", "error": str(e)} @api.post("/generate") async def generate(req: Request): p = await req.json() try: return await run_in_threadpool(generate_pack, str(p.get("subject", ""))[:120]) except Exception as e: return {"error": str(e)} @api.post("/publish") async def publish_post(req: Request): p = await req.json() try: wid = await run_in_threadpool(community.publish, p.get("pack"), p.get("by")) return {"ok": bool(wid), "id": wid} except Exception as e: return {"ok": False, "error": str(e)} @api.get("/puzzles") async def puzzles_list(req: Request): try: sort = "top" if req.query_params.get("sort") == "top" else "new" q = req.query_params.get("q", "") return {"enabled": community.ENABLED, "rows": await run_in_threadpool(community.latest, 24, sort, q)} except Exception as e: return {"enabled": False, "rows": [], "error": str(e)} @api.get("/puzzle/{wid}") async def puzzle_get(wid: str): try: pack = await run_in_threadpool(community.get, wid) return pack if pack else {"error": "not found"} except Exception as e: return {"error": str(e)} @api.post("/like") async def like_post(req: Request): p = await req.json() try: likes = await run_in_threadpool(community.like, p.get("id")) return {"ok": True, "likes": likes} except Exception as e: return {"ok": False, "error": str(e)} @api.get("/times") async def times_get(req: Request): try: pid = req.query_params.get("p", "") pieces = int(req.query_params.get("n", "24")) rows = await run_in_threadpool(leaderboard.top, pid, pieces, 20) out = {"enabled": leaderboard.ENABLED, "rows": rows} my = req.query_params.get("my") # caller's time -> its rank on this board if my is not None: rank, total = await run_in_threadpool(leaderboard.rank_of, pid, pieces, int(my)) out.update(myRank=rank, total=total) return out except Exception as e: return {"enabled": False, "rows": [], "error": str(e)} @api.post("/times") async def times_post(req: Request): p = await req.json() try: ok = await run_in_threadpool( leaderboard.submit, p.get("p"), p.get("pieces", 0), p.get("name"), p.get("ms", 0), p.get("moves", 0), bool(p.get("rot"))) return {"ok": ok} except Exception as e: return {"ok": False, "error": str(e)} class NoCacheStatic(StaticFiles): """no-cache (NOT no-store): browsers revalidate by ETag — cheap 304s normally, instantly fresh after every push. Stale game files have bitten us repeatedly.""" def file_response(self, *args, **kwargs): resp = super().file_response(*args, **kwargs) resp.headers["Cache-Control"] = "no-cache" return resp api.mount("/game", NoCacheStatic(directory=STATIC, html=True), name="game") # /game avoids gradio's own /static def generate_puzzle(subject: str) -> dict: """Puzzle generation as a GRADIO API endpoint (/gradio_api/call/generate_puzzle). On ZeroGPU the @spaces.GPU call must run inside a Gradio request for correct quota attribution — the game calls this path first, /generate is the fallback.""" try: return generate_pack(str(subject or "")[:120]) except Exception as e: return {"error": str(e)} # The game iframe can't attach ZeroGPU's visitor token to raw fetches — such calls # arrive identity-less and get ZERO quota ("0s left"), for every account. So generation # runs as a REAL gradio event on the OUTER page (whose native client carries the token), # and the game iframe drives it via postMessage. Hidden components below are the relay. def _generate_puzzle_str(subject): return json.dumps(generate_puzzle(subject)) _BRIDGE_JS = """ () => { window.addEventListener('message', (ev) => { const d = ev.data || {}; const frame = document.getElementById('pz-frame'); const reply = (msg) => { try { ((frame && frame.contentWindow) || ev.source).postMessage(msg, '*'); } catch (e) {} }; if (d.type === 'ajp-ping') { reply({ type: 'ajp-pong' }); return; } // handshake: lets the game detect the bridge fast if (d.type !== 'ajp-gen') return; const inEl = document.querySelector('#gw-in textarea, #gw-in input'); const outEl = document.querySelector('#gw-out textarea, #gw-out input'); const btn = document.querySelector('#gw-btn button') || document.querySelector('#gw-btn'); if (!inEl || !outEl || !btn) { reply({ type: 'ajp-gen-result', error: 'bridge components missing' }); return; } const setVal = (el, v) => { const proto = el.tagName === 'TEXTAREA' ? window.HTMLTextAreaElement.prototype : window.HTMLInputElement.prototype; Object.getOwnPropertyDescriptor(proto, 'value').set.call(el, v); el.dispatchEvent(new Event('input', { bubbles: true })); }; setVal(outEl, ''); setVal(inEl, String(d.subject || '')); setTimeout(() => btn.click(), 80); const t0 = Date.now(); const iv = setInterval(() => { if (outEl.value) { clearInterval(iv); let parsed = null; try { parsed = JSON.parse(outEl.value); } catch (e) {} reply({ type: 'ajp-gen-result', data: parsed, error: parsed ? null : 'bad payload' }); } else if (Date.now() - t0 > 240000) { clearInterval(iv); reply({ type: 'ajp-gen-result', error: 'generation timed out' }); } }, 400); }); } """ # ---------------- the Gradio app: custom UI = the embedded canvas game ---------------- with gr.Blocks(title="AI Puzzle Maker", analytics_enabled=False) as demo: gr.api(generate_puzzle, api_name="generate_puzzle") with gr.Column(elem_id="gw-wrap"): # parked off-screen via CSS — NOT visible=False (must stay in the DOM) gw_in = gr.Textbox(label="subject", elem_id="gw-in") gw_btn = gr.Button("generate", elem_id="gw-btn") gw_out = gr.Textbox(label="result", elem_id="gw-out") gw_btn.click(_generate_puzzle_str, inputs=gw_in, outputs=gw_out) demo.load(fn=None, inputs=None, outputs=None, js=_BRIDGE_JS) gr.HTML( "" '' ) if __name__ == "__main__": moderation.prewarm() # warm the Modal safety endpoint in the background (no-op when disabled) if os.environ.get("SPACE_ID"): # On a Gradio-SDK Space (required for ZeroGPU) the SDK expects demo.launch() # to own the server — running our own uvicorn makes gradio try to bind a # second port and the container crash-loops. So: launch gradio first, then # attach the game's routes to ITS FastAPI app. demo.launch(prevent_thread_lock=True, ssr_mode=False) # SSR blank-renders custom-HTML-only apps demo.app.post("/quip")(quip) demo.app.post("/generate")(generate) demo.app.post("/publish")(publish_post) demo.app.get("/puzzles")(puzzles_list) demo.app.get("/puzzle/{wid}")(puzzle_get) demo.app.post("/like")(like_post) demo.app.get("/times")(times_get) demo.app.post("/times")(times_post) demo.app.mount("/game", NoCacheStatic(directory=STATIC, html=True), name="game") demo.block_thread() else: # Local dev: FastAPI-first (unchanged) import uvicorn app = gr.mount_gradio_app(api, demo, path="/") uvicorn.run(app, host="0.0.0.0", port=int(os.environ.get("PORT", 7860)))