194 lines
7.2 KiB
Python
194 lines
7.2 KiB
Python
import json
|
|
import os
|
|
import re
|
|
import secrets
|
|
import sqlite3
|
|
import time
|
|
import uuid
|
|
from pathlib import Path
|
|
|
|
from fastapi import Body, FastAPI, Header, HTTPException, Response
|
|
from fastapi.responses import FileResponse
|
|
from fastapi.staticfiles import StaticFiles
|
|
|
|
from .export import to_docx, to_pdf
|
|
|
|
BASE = Path(__file__).resolve().parent.parent
|
|
RULES_PATH = Path(os.environ.get("AW_RULES", BASE / "data" / "rules.json"))
|
|
DB_PATH = os.environ.get("AW_DB", str(BASE / "aw.db"))
|
|
STATIC = Path(os.environ.get("AW_STATIC", BASE / "static"))
|
|
MAX_BODY = 200_000 # bytes per character
|
|
MAX_PRESETS = 200
|
|
|
|
app = FastAPI(title="Arcane Wasteland Character Creator")
|
|
|
|
|
|
def load_rules() -> dict:
|
|
# Re-read on every request so rules.json can be edited without a restart.
|
|
return json.loads(RULES_PATH.read_text(encoding="utf-8"))
|
|
|
|
|
|
def db() -> sqlite3.Connection:
|
|
Path(DB_PATH).parent.mkdir(parents=True, exist_ok=True)
|
|
con = sqlite3.connect(DB_PATH)
|
|
con.row_factory = sqlite3.Row
|
|
con.execute("""CREATE TABLE IF NOT EXISTS characters (
|
|
id TEXT PRIMARY KEY, token TEXT NOT NULL, data TEXT NOT NULL,
|
|
created REAL NOT NULL, updated REAL NOT NULL)""")
|
|
con.execute("""CREATE TABLE IF NOT EXISTS presets (
|
|
id TEXT PRIMARY KEY, token TEXT NOT NULL, data TEXT NOT NULL,
|
|
created REAL NOT NULL, updated REAL NOT NULL)""")
|
|
return con
|
|
|
|
|
|
def custom_presets() -> list:
|
|
with db() as con:
|
|
rows = con.execute("SELECT id, data FROM presets ORDER BY created").fetchall()
|
|
return [{"id": r["id"], **json.loads(r["data"]), "custom": True} for r in rows]
|
|
|
|
|
|
def all_rules() -> dict:
|
|
"""Rules from rules.json plus the presets created in the app."""
|
|
r = load_rules()
|
|
r["presets"] = [{**p, "builtin": True} for p in r.get("presets", [])] + custom_presets()
|
|
return r
|
|
|
|
|
|
def clean_preset(body: dict, rules: dict) -> dict:
|
|
"""Validates a preset: name, description and only known creation numbers within their limits."""
|
|
name = str(body.get("name", "")).strip()[:60]
|
|
if not name:
|
|
raise HTTPException(422, "A preset needs a name")
|
|
tun = {t["key"]: t for t in rules.get("tunables", [])}
|
|
creation = {}
|
|
for k, v in (body.get("creation") or {}).items():
|
|
if k not in tun:
|
|
raise HTTPException(422, f"Unknown value: {k}")
|
|
if isinstance(v, bool) or not isinstance(v, int) or not tun[k]["min"] <= v <= tun[k]["max"]:
|
|
raise HTTPException(422, f"Value out of range: {tun[k]['label']}")
|
|
if v != rules["creation"].get(k):
|
|
creation[k] = v
|
|
return {"name": name, "desc": str(body.get("desc", "")).strip()[:300], "creation": creation}
|
|
|
|
|
|
def check_size(ch: dict):
|
|
if len(json.dumps(ch)) > MAX_BODY:
|
|
raise HTTPException(413, "Character too large")
|
|
|
|
|
|
@app.get("/api/rules")
|
|
def get_rules():
|
|
return all_rules()
|
|
|
|
|
|
@app.post("/api/presets")
|
|
def create_preset(body: dict = Body(...)):
|
|
data = clean_preset(body, load_rules())
|
|
with db() as con:
|
|
if con.execute("SELECT COUNT(*) FROM presets").fetchone()[0] >= MAX_PRESETS:
|
|
raise HTTPException(409, "Too many presets on this server")
|
|
pid, token, now = "u-" + uuid.uuid4().hex[:8], secrets.token_urlsafe(16), time.time()
|
|
con.execute("INSERT INTO presets VALUES (?,?,?,?,?)", (pid, token, json.dumps(data), now, now))
|
|
return {"id": pid, "token": token}
|
|
|
|
|
|
@app.put("/api/presets/{pid}")
|
|
def update_preset(pid: str, body: dict = Body(...), x_edit_token: str = Header(default="")):
|
|
data = clean_preset(body, load_rules())
|
|
with db() as con:
|
|
row = con.execute("SELECT token FROM presets WHERE id=?", (pid,)).fetchone()
|
|
if not row:
|
|
raise HTTPException(404, "Preset not found")
|
|
if not secrets.compare_digest(row["token"], x_edit_token):
|
|
raise HTTPException(403, "No write access to this preset")
|
|
con.execute("UPDATE presets SET data=?, updated=? WHERE id=?", (json.dumps(data), time.time(), pid))
|
|
return {"ok": True}
|
|
|
|
|
|
@app.delete("/api/presets/{pid}")
|
|
def delete_preset(pid: str, x_edit_token: str = Header(default="")):
|
|
with db() as con:
|
|
row = con.execute("SELECT token FROM presets WHERE id=?", (pid,)).fetchone()
|
|
if not row:
|
|
raise HTTPException(404, "Preset not found")
|
|
if not secrets.compare_digest(row["token"], x_edit_token):
|
|
raise HTTPException(403, "No write access to this preset")
|
|
con.execute("DELETE FROM presets WHERE id=?", (pid,))
|
|
return {"ok": True}
|
|
|
|
|
|
@app.post("/api/characters")
|
|
def create_character(ch: dict = Body(...)):
|
|
check_size(ch)
|
|
cid, token, now = uuid.uuid4().hex[:12], secrets.token_urlsafe(16), time.time()
|
|
with db() as con:
|
|
con.execute("INSERT INTO characters VALUES (?,?,?,?,?)", (cid, token, json.dumps(ch), now, now))
|
|
return {"id": cid, "token": token}
|
|
|
|
|
|
@app.get("/api/characters/{cid}")
|
|
def read_character(cid: str):
|
|
with db() as con:
|
|
row = con.execute("SELECT data FROM characters WHERE id=?", (cid,)).fetchone()
|
|
if not row:
|
|
raise HTTPException(404, "Character not found")
|
|
return json.loads(row["data"])
|
|
|
|
|
|
@app.put("/api/characters/{cid}")
|
|
def update_character(cid: str, ch: dict = Body(...), x_edit_token: str = Header(default="")):
|
|
check_size(ch)
|
|
with db() as con:
|
|
row = con.execute("SELECT token FROM characters WHERE id=?", (cid,)).fetchone()
|
|
if not row:
|
|
raise HTTPException(404, "Character not found")
|
|
if not secrets.compare_digest(row["token"], x_edit_token):
|
|
raise HTTPException(403, "No write access to this character")
|
|
con.execute("UPDATE characters SET data=?, updated=? WHERE id=?", (json.dumps(ch), time.time(), cid))
|
|
return {"ok": True}
|
|
|
|
|
|
@app.delete("/api/characters/{cid}")
|
|
def delete_character(cid: str, x_edit_token: str = Header(default="")):
|
|
with db() as con:
|
|
row = con.execute("SELECT token FROM characters WHERE id=?", (cid,)).fetchone()
|
|
if not row:
|
|
raise HTTPException(404, "Character not found")
|
|
if not secrets.compare_digest(row["token"], x_edit_token):
|
|
raise HTTPException(403, "No write access to this character")
|
|
con.execute("DELETE FROM characters WHERE id=?", (cid,))
|
|
return {"ok": True}
|
|
|
|
|
|
def _filename(ch: dict, ext: str) -> str:
|
|
base = re.sub(r"[^A-Za-z0-9_-]+", "_", ch.get("name") or "Character").strip("_") or "Character"
|
|
return f"{base}.{ext}"
|
|
|
|
|
|
@app.post("/api/export/{fmt}")
|
|
def export(fmt: str, ch: dict = Body(...)):
|
|
check_size(ch)
|
|
rules = all_rules()
|
|
if fmt == "docx":
|
|
data = to_docx(ch, rules)
|
|
mime = "application/vnd.openxmlformats-officedocument.wordprocessingml.document"
|
|
elif fmt == "pdf":
|
|
data = to_pdf(ch, rules)
|
|
mime = "application/pdf"
|
|
else:
|
|
raise HTTPException(404, "Unknown format")
|
|
return Response(data, media_type=mime,
|
|
headers={"Content-Disposition": f'attachment; filename="{_filename(ch, fmt)}"'})
|
|
|
|
|
|
# Serve the built frontend with an SPA fallback
|
|
if STATIC.is_dir():
|
|
app.mount("/assets", StaticFiles(directory=STATIC / "assets"), name="assets")
|
|
|
|
@app.get("/{path:path}")
|
|
def spa(path: str):
|
|
f = (STATIC / path).resolve()
|
|
if path and f.is_file() and STATIC in f.parents:
|
|
return FileResponse(f)
|
|
return FileResponse(STATIC / "index.html")
|