247 lines
8.9 KiB
Python
247 lines
8.9 KiB
Python
"""SCENEGOD server — page + assets + scenes + render pipeline.
|
|
|
|
Single-file FastAPI app (MESHGOD convention). A plain folder tree is the asset
|
|
database (no index file to drift): assets/{characters,animations,props,
|
|
backdrops,audio}/<pack>/name.ext. Files sharing dir+stem = one entry, many
|
|
formats. See PLAN.md §4.3 for the HTTP contract.
|
|
|
|
uvicorn scenegod.server:app --port 8020 # binds 127.0.0.1
|
|
"""
|
|
import json
|
|
import os
|
|
import re
|
|
import shutil
|
|
import time
|
|
import uuid
|
|
from pathlib import Path
|
|
|
|
from fastapi import FastAPI, HTTPException, Request
|
|
from fastapi.responses import FileResponse, HTMLResponse, JSONResponse, Response
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
WEB = Path(__file__).resolve().parent / "web" # scenegod/web (PLAN §2)
|
|
ASSETS = Path(os.environ.get("SCENEGOD_ASSETS", ROOT / "assets")).resolve()
|
|
SCENES = Path(os.environ.get("SCENEGOD_SCENES", ROOT / "scenes")).resolve()
|
|
RENDERS = Path(os.environ.get("SCENEGOD_RENDERS", ROOT / "renders")).resolve()
|
|
|
|
MODEL_EXTS = {"glb", "gltf", "fbx", "obj", "bvh"}
|
|
IMG_EXTS = {"jpg", "jpeg", "png", "webp"}
|
|
AUDIO_EXTS = {"wav", "mp3"}
|
|
ASSET_EXTS = MODEL_EXTS | IMG_EXTS | AUDIO_EXTS
|
|
# primary ext per category (others sharing a stem are thumbnails)
|
|
CATEGORIES = {"characters": MODEL_EXTS, "animations": MODEL_EXTS, "props": MODEL_EXTS,
|
|
"backdrops": IMG_EXTS, "audio": AUDIO_EXTS}
|
|
MEDIA_TYPES = {
|
|
"glb": "model/gltf-binary", "gltf": "model/gltf+json", "fbx": "application/octet-stream",
|
|
"obj": "text/plain", "bvh": "text/plain", "jpg": "image/jpeg", "jpeg": "image/jpeg",
|
|
"png": "image/png", "webp": "image/webp", "wav": "audio/wav", "mp3": "audio/mpeg",
|
|
}
|
|
|
|
SCENES.mkdir(parents=True, exist_ok=True)
|
|
RENDERS.mkdir(parents=True, exist_ok=True)
|
|
HAVE_FFMPEG = shutil.which("ffmpeg") is not None
|
|
|
|
app = FastAPI(title="SCENEGOD")
|
|
|
|
|
|
# ---- path guard (ship-check rule: every path-taking endpoint uses this) ----
|
|
def safe_under(root: Path, relpath: str) -> Path:
|
|
"""Resolve relpath under root; reject traversal, absolute, symlink escape."""
|
|
full = (root / relpath).resolve()
|
|
if full != root and root not in full.parents:
|
|
raise HTTPException(400, "bad path")
|
|
return full
|
|
|
|
|
|
# ---- assets ----
|
|
_tree_cache = {"t": 0.0, "val": None}
|
|
|
|
|
|
def scan_assets() -> dict:
|
|
"""Walk each category dir, group files by (dir, stem) → one entry w/ formats.
|
|
Live scan (hundreds of files); cached 5s in /assets/tree."""
|
|
out = {c: [] for c in CATEGORIES}
|
|
for cat, primary in CATEGORIES.items():
|
|
base = ASSETS / cat
|
|
if not base.is_dir():
|
|
continue
|
|
groups: dict[str, dict] = {}
|
|
for dirpath, dirs, files in os.walk(base):
|
|
dirs[:] = [d for d in dirs if not d.startswith(".")]
|
|
for f in sorted(files):
|
|
stem, dot, ext = f.rpartition(".")
|
|
ext = ext.lower()
|
|
if f.startswith(".") or not dot or ext not in ASSET_EXTS:
|
|
continue
|
|
rel = os.path.relpath(os.path.join(dirpath, f), ASSETS).replace(os.sep, "/")
|
|
key = os.path.relpath(os.path.join(dirpath, stem), ASSETS) # dir+stem
|
|
g = groups.setdefault(key, {"name": stem, "path": None, "formats": [], "thumb": None})
|
|
if ext not in primary:
|
|
if ext in IMG_EXTS:
|
|
g["thumb"] = rel
|
|
continue
|
|
g["formats"].append(ext)
|
|
if g["path"] is None or ext == "glb": # prefer glb, else first
|
|
g["path"] = rel
|
|
entries = [g for g in groups.values() if g["formats"]] # thumb-only stems aren't assets
|
|
out[cat] = sorted(entries, key=lambda g: g["name"].lower())
|
|
return out
|
|
|
|
|
|
@app.get("/assets/tree")
|
|
def assets_tree():
|
|
if not ASSETS.is_dir():
|
|
raise HTTPException(500, f"asset root missing: {ASSETS} (set SCENEGOD_ASSETS)")
|
|
now = time.monotonic()
|
|
if _tree_cache["val"] is None or now - _tree_cache["t"] > 5.0:
|
|
_tree_cache["val"] = scan_assets()
|
|
_tree_cache["t"] = now
|
|
return _tree_cache["val"]
|
|
|
|
|
|
@app.get("/assets/file")
|
|
def assets_file(path: str):
|
|
full = safe_under(ASSETS, path)
|
|
if not full.is_file():
|
|
raise HTTPException(404, "not found")
|
|
ext = full.suffix.lstrip(".").lower()
|
|
return FileResponse(full, media_type=MEDIA_TYPES.get(ext, "application/octet-stream"))
|
|
|
|
|
|
# ---- scenes ----
|
|
SLUG = re.compile(r"[^a-z0-9_-]")
|
|
|
|
|
|
def slugify(name: str) -> str:
|
|
s = SLUG.sub("-", name.lower()).strip("-")
|
|
if not s:
|
|
raise HTTPException(422, "empty scene name after slugify")
|
|
return s
|
|
|
|
|
|
def validate_scene(s: dict) -> list[str]:
|
|
"""Return list of violations (empty = valid). Stdlib only, no jsonschema."""
|
|
errs = []
|
|
if s.get("version") != 1:
|
|
errs.append(f"version must be 1, got {s.get('version')!r}")
|
|
ents = s.get("entities")
|
|
if not isinstance(ents, list):
|
|
return errs + ["entities must be a list"]
|
|
ids, cam_ids = set(), set()
|
|
for i, e in enumerate(ents):
|
|
eid = e.get("id")
|
|
if eid in ids:
|
|
errs.append(f"entity[{i}]: duplicate id {eid!r}")
|
|
ids.add(eid)
|
|
if e.get("kind") == "camera":
|
|
cam_ids.add(eid)
|
|
tracks = e.get("tracks") or {}
|
|
for tk in ("transform", "params"):
|
|
keys = tracks.get(tk) or []
|
|
ts = [k.get("t") for k in keys]
|
|
if ts != sorted(ts):
|
|
errs.append(f"entity {eid!r}: track {tk} not sorted by t")
|
|
clips = sorted(tracks.get("clips") or [], key=lambda c: c.get("start", 0))
|
|
for a, b in zip(clips, clips[1:]):
|
|
end = a.get("start", 0) + (a.get("out", 0) - a.get("in", 0)) * a.get("loop", 1)
|
|
if b.get("start", 0) < end - a.get("fade", 0) - 1e-6:
|
|
errs.append(f"entity {eid!r}: clips overlap beyond fade "
|
|
f"(block ends {end:.3f}, next starts {b.get('start')})")
|
|
for cut in s.get("cameraCuts") or []:
|
|
if cut.get("camera") not in cam_ids:
|
|
errs.append(f"cameraCut references non-camera entity {cut.get('camera')!r}")
|
|
return errs
|
|
|
|
|
|
@app.get("/scenes")
|
|
def scenes_list():
|
|
out = []
|
|
for p in sorted(SCENES.glob("*.json")):
|
|
try:
|
|
dur = json.loads(p.read_text()).get("duration")
|
|
except Exception:
|
|
dur = None
|
|
out.append({"name": p.stem, "mtime": int(p.stat().st_mtime), "duration": dur})
|
|
return out
|
|
|
|
|
|
@app.get("/scenes/{name}")
|
|
def scene_get(name: str):
|
|
p = safe_under(SCENES, slugify(name) + ".json")
|
|
if not p.is_file():
|
|
raise HTTPException(404, "not found")
|
|
return json.loads(p.read_text())
|
|
|
|
|
|
@app.post("/scenes/{name}")
|
|
async def scene_save(name: str, request: Request):
|
|
body = await request.body()
|
|
try:
|
|
scene = json.loads(body)
|
|
except json.JSONDecodeError as e:
|
|
raise HTTPException(400, f"invalid JSON: {e}")
|
|
errs = validate_scene(scene)
|
|
if errs:
|
|
return JSONResponse({"errors": errs}, status_code=422)
|
|
p = safe_under(SCENES, slugify(name) + ".json")
|
|
tmp = p.with_suffix(".json.tmp")
|
|
tmp.write_text(json.dumps(scene, indent=2))
|
|
tmp.replace(p) # atomic
|
|
return {"ok": True, "name": p.stem}
|
|
|
|
|
|
# ---- render (M3 — design landed, encode stubbed) ----
|
|
@app.post("/render/begin")
|
|
async def render_begin(request: Request):
|
|
if not HAVE_FFMPEG:
|
|
raise HTTPException(503, "ffmpeg not found on server")
|
|
rid = uuid.uuid4().hex
|
|
(RENDERS / rid / "frames").mkdir(parents=True)
|
|
(RENDERS / rid / "meta.json").write_text((await request.body()).decode() or "{}")
|
|
return {"renderId": rid}
|
|
|
|
|
|
@app.post("/render/{rid}/frame/{n}")
|
|
async def render_frame(rid: str, n: int, request: Request):
|
|
frames = safe_under(RENDERS, rid) / "frames"
|
|
if not frames.is_dir():
|
|
raise HTTPException(404, "unknown renderId")
|
|
(frames / f"{n:06d}.png").write_bytes(await request.body())
|
|
return {"ok": True}
|
|
|
|
|
|
@app.post("/render/{rid}/end")
|
|
async def render_end(rid: str):
|
|
raise HTTPException(501, "encode not implemented until M3")
|
|
|
|
|
|
@app.get("/render/{rid}/status")
|
|
def render_status(rid: str):
|
|
raise HTTPException(501, "not implemented until M3")
|
|
|
|
|
|
@app.get("/render/{rid}/out.mp4")
|
|
def render_out(rid: str):
|
|
raise HTTPException(501, "not implemented until M3")
|
|
|
|
|
|
# ---- statics (last: catch-all /web is defined after API routes) ----
|
|
@app.get("/", response_class=HTMLResponse)
|
|
def index():
|
|
f = WEB / "index.html"
|
|
html = f.read_text() if f.is_file() else "<h1>SCENEGOD</h1><p>Lane A page not landed yet.</p>"
|
|
return HTMLResponse(html, headers={"Cache-Control": "no-store"}) # MESHGOD stale-cache lesson
|
|
|
|
|
|
@app.get("/web/{path:path}")
|
|
def web_static(path: str):
|
|
full = safe_under(WEB, path)
|
|
if not full.is_file():
|
|
raise HTTPException(404, "not found")
|
|
return FileResponse(full, headers={"Cache-Control": "max-age=60"})
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import uvicorn
|
|
uvicorn.run(app, host="127.0.0.1", port=int(os.environ.get("SCENEGOD_PORT", 8020)))
|