"""
LLM Cinema — CUSTOM frontend on gr.Server (Gradio's FastAPI API engine).
This is the "off-brand" build: instead of Gradio components, we serve our OWN
HTML/CSS/JS cinema and stream frames over Server-Sent Events from FastAPI routes
on a gr.Server instance. The film engine (movies + render) is reused as-is.
Run: python server_app.py
Deploy: HF Space (set app_file: server_app.py). Same CLAUDEMOVIES_LLM_* secrets.
"""
import html
import json
import os
import re
import threading
import time
import uuid
import gradio as gr
from fastapi.responses import FileResponse, HTMLResponse, JSONResponse, StreamingResponse
import movies
import render
W, H = render.W, render.H
HERE = os.path.dirname(__file__)
SHOW = os.path.join(HERE, "showcase")
# ── content filter (profanity / slurs / hate) ──
_BANNED = re.compile(r"\b(" + "|".join([
"fuck", "fucks", "fucking", "fucker", "shit", "bitch", "cunt", "asshole", "bastard",
"dick", "piss", "slut", "whore", "nigger", "nigga", "faggot", "fag", "kike", "spic",
"chink", "gook", "wetback", "tranny", "retard", "coon", "dyke", "paki", "beaner",
"heil hitler", "sieg heil", "white power", "kkk", "gas the jews",
]) + r")\b", re.I)
def _blocked(t):
return bool(_BANNED.search(t or ""))
_MAXLEN = 60
# common prompt-injection / jailbreak patterns — neutralised so nobody can steer the model
_INJECT = re.compile(
r"(ignore\s+(all\s+|the\s+)?(previous|prior|above|earlier)"
r"|disregard\s+(all|the|previous|prior)"
r"|system\s+prompt|you\s+are\s+now|act\s+as\s+(a|an|if)"
r"|jailbreak|developer\s+mode|override\s+(the|your)"
r"|forget\s+(everything|all|the)|new\s+instructions)", re.I)
def _clean_concept(s):
"""Authoritative server-side sanitiser: strip control chars / markup, collapse whitespace,
and HARD-cap to 60 chars so a single short film idea is all the model ever receives."""
s = re.sub(r"[\x00-\x1f\x7f]", " ", s or "") # control chars, newlines, tabs
s = "".join(ch for ch in s if ch.isprintable())
s = re.sub(r"[<>{}`\\]", "", s) # markup / structural chars
s = re.sub(r"\s+", " ", s).strip()
return s[:_MAXLEN]
def _films(folder):
out = {}
order = ["knight.json", "paper-boat.json", "rain-cat.json", "troll-dance.json",
"snail-race.json", "ghost.json", "bread-dragon.json"]
paths = sorted(__import__("glob").glob(os.path.join(folder, "*.json")),
key=lambda p: (order.index(os.path.basename(p)) if os.path.basename(p) in order else 99,
os.path.basename(p)))
for p in paths:
try:
out[json.load(open(p)).get("title") or os.path.basename(p)[:-5]] = p
except Exception:
continue
return out
def grid_to_html(grid, title=""):
rows = []
for r in range(H):
cells, row, i = grid[r], "", 0
while i < W:
col, j, seg = cells[i][1], i, ""
while j < W and cells[j][1] == col:
seg += cells[j][0]
j += 1
hexc = "#%02x%02x%02x" % col if isinstance(col, tuple) else "#7df9a6"
esc = html.escape(seg)
row += f"{esc}" if seg.strip() else esc
i = j
rows.append(row)
bar = (f"
now_playing: {html.escape(title)}
")
return f"{bar}
{chr(10).join(rows)}
"
def _tint(grid):
"""The frame's lit colour and 'energy' — a saturation/brightness-weighted average of the
drawn cells, so the audience glow becomes a low-opacity MIRROR of the action on screen.
Energy is keyed to how much is lit, so sparse frames (the title/credit cards) glow ~0."""
tr = tg = tb = tw = 0.0
lit = 0
for row in grid:
for cell in row:
ch = cell[0]
if ch == " " or ch == "":
continue
r, g, b = cell[1]
lit += 1
mx, mn = max(r, g, b), min(r, g, b)
lum = 0.2126 * r + 0.7152 * g + 0.0722 * b
sat = (mx - mn) / mx if mx else 0.0
w = lum * (0.35 + sat) # vivid + bright cells lead the colour
tr += r * w; tg += g * w; tb += b * w; tw += w
cov = lit / float(W * H)
energy = max(0.0, min(1.0, (cov - 0.10) * 6.0)) # cards are sparse -> ~0 -> no glow
if tw <= 0:
return [150, 180, 235], 0.0
return [int(tr / tw), int(tg / tw), int(tb / tw)], round(energy, 3)
def _payload(grid, title=""):
tint, energy = _tint(grid)
return {"html": grid_to_html(grid, title), "tint": tint, "lvl": energy}
def _screen(msg, label="ready"):
bar = f"
{html.escape(label)}
"
return f"{bar}
{html.escape(msg)}
"
def _sse(frames):
for fr in frames:
payload = fr if isinstance(fr, dict) else {"html": fr}
yield "data: " + json.dumps(payload) + "\n\n"
yield "data: " + json.dumps({"done": True}) + "\n\n"
# ── the custom frontend: a real cinema-hall photo as the backdrop, the ASCII film
# projected onto the screen, and all controls docked along the bottom ──
PAGE = r"""
LLM Cinema
presented by conductor creative labs
LLM CINEMA
▶ LLM CINEMA
"""
server = gr.Server(title="LLM Cinema")
@server.get("/")
def index():
return HTMLResponse(PAGE)
@server.get("/static/cinema.png")
def bg_image():
return FileResponse(os.path.join(HERE, "static", "cinema.png"))
@server.get("/static/ansifont.json")
def ansifont():
return FileResponse(os.path.join(HERE, "static", "ansifont.json"), media_type="application/json")
@server.get("/static/share.png")
def share_image():
return FileResponse(os.path.join(HERE, "static", "share.png"), media_type="image/png")
@server.get("/api/gallery")
def gallery():
return JSONResponse(list(_films(SHOW).keys()))
@server.get("/api/play")
def play(name: str):
path = _films(SHOW).get(name)
if not path:
return JSONResponse({"error": "not found"}, status_code=404)
spec = json.load(open(path))
def frames():
for grid in render.iter_movie_frames(spec):
yield _payload(grid, spec.get("title", ""))
time.sleep(movies.FRAME_MS / 1000)
return StreamingResponse(_sse(frames()), media_type="text/event-stream")
@server.get("/api/make")
def make(concept: str):
concept = _clean_concept(concept) # sanitise + hard 60-char cap (authoritative)
def frames():
if not concept:
yield _screen("Type a short film idea to begin.", "ready")
return
if _blocked(concept) or _INJECT.search(concept):
yield _screen("Let's keep it friendly — try a kind, fun film idea.", "blocked")
return
result = {}
def work():
try:
result["spec"] = movies.direct(concept)
except Exception as e:
result["error"] = f"{type(e).__name__}: {e}"
th = threading.Thread(target=work, daemon=True)
th.start()
t0 = time.time()
while th.is_alive():
# heartbeats keep the SSE alive while the client plays the opening credits
yield {"phase": "writing", "t": int(time.time() - t0)}
th.join(timeout=0.7)
if result.get("error") or not (result.get("spec") or {}).get("shots"):
yield _screen("Could not reach the model. Please try again.", "error")
return
spec = result["spec"]
for grid in render.iter_movie_frames(spec):
yield _payload(grid, spec.get("title", ""))
time.sleep(movies.FRAME_MS / 1000)
return StreamingResponse(_sse(frames()), media_type="text/event-stream")
if __name__ == "__main__":
# HF Spaces runs this file as a script and injects GRADIO_SERVER_NAME/PORT.
server.launch(server_name=os.environ.get("GRADIO_SERVER_NAME", "0.0.0.0"),
server_port=int(os.environ.get("GRADIO_SERVER_PORT", "7860")))
