NPC editor, NPC tokens from library, admin management

- NPC library (own + shared) built on the character wizard with a free NPC preset
- NPC tokens in map editor and lobby, DM sheet on token click, DM checks for NPCs
- Admins can promote/demote other users

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HJhXuLD2Z6hv3Wqeq9UL9K
This commit is contained in:
Claude
2026-10-23 21:36:09 +00:00
parent 57f262b6aa
commit 1db8a45cae
13 changed files with 364 additions and 86 deletions
+115 -28
View File
@@ -103,6 +103,10 @@ def db() -> sqlite3.Connection:
("lobby_members", "char_mod", "INTEGER NOT NULL DEFAULT 0")):
if col not in [r["name"] for r in con.execute(f"PRAGMA table_info({table})")]:
con.execute(f"ALTER TABLE {table} ADD COLUMN {col} {ddl}")
if "is_admin" not in [r["name"] for r in con.execute("PRAGMA table_info(users)")]:
con.execute("ALTER TABLE users ADD COLUMN is_admin INTEGER NOT NULL DEFAULT 0")
if "npc" not in [r["name"] for r in con.execute("PRAGMA table_info(characters)")]:
con.execute("ALTER TABLE characters ADD COLUMN npc INTEGER NOT NULL DEFAULT 0")
maps_mod.schema(con)
sounds_mod.schema(con)
for table in ("characters", "presets"): # older databases: add the owner column
@@ -188,14 +192,21 @@ async def security(request: Request, call_next):
return resp
def is_admin(user: dict) -> bool:
"""Admins (rules editor): the names in AW_ADMINS, or without that setting the very first account."""
def admin_ids(con) -> set:
"""Admins: the names in AW_ADMINS, accounts an admin promoted, and (when there is neither) the very first account."""
names = {n.strip().lower() for n in os.environ.get("AW_ADMINS", "").split(",") if n.strip()}
if names:
return user["username"].lower() in names
with db() as con:
ids = {r["id"] for r in con.execute("SELECT id, username FROM users") if r["username"].lower() in names}
ids |= {r["id"] for r in con.execute("SELECT id FROM users WHERE is_admin=1")}
if not ids:
first = con.execute("SELECT id FROM users ORDER BY id LIMIT 1").fetchone()
return bool(first) and first["id"] == user["id"]
if first:
ids.add(first["id"])
return ids
def is_admin(user: dict) -> bool:
with db() as con:
return user["id"] in admin_ids(con)
def current_user(request: Request) -> dict:
@@ -366,23 +377,35 @@ def _name(data: str) -> str:
return json.loads(data).get("name") or "Nameless"
def _entry(r, shares=None, owner=None) -> dict:
ch = json.loads(r["data"])
e = {"id": r["id"], "name": ch.get("name") or "Nameless", "updated": r["updated"], "portrait": portrait_mod.tag(ch.get("portrait"))}
if r["npc"]:
e["kind"] = str(ch.get("npc_kind") or "")
e["size"] = ch.get("token_size", 1)
if shares is not None:
e["shared_with"] = shares.get(r["id"], [])
if owner is not None:
e["owner"] = owner or "?"
return e
@api.get("/characters")
def my_characters(user: dict = Depends(current_user)):
def my_characters(npc: int = 0, user: dict = Depends(current_user)):
"""Your characters and the ones shared with you. With ?npc=1 the NPC library (NPCs, enemies, creatures) instead."""
flag = 1 if npc else 0
with db() as con:
rows = con.execute("SELECT id, data, updated FROM characters WHERE owner_id=? ORDER BY updated DESC",
(user["id"],)).fetchall()
rows = con.execute("SELECT id, data, updated, npc FROM characters WHERE owner_id=? AND npc=? ORDER BY updated DESC",
(user["id"], flag)).fetchall()
shares: dict = {}
for r in con.execute("""SELECT s.character_id, u.username FROM shares s JOIN users u ON u.id = s.user_id
JOIN characters c ON c.id = s.character_id WHERE c.owner_id=? ORDER BY u.username""",
(user["id"],)):
JOIN characters c ON c.id = s.character_id WHERE c.owner_id=? AND c.npc=? ORDER BY u.username""",
(user["id"], flag)):
shares.setdefault(r["character_id"], []).append(r["username"])
shared = con.execute("""SELECT c.id, c.data, c.updated, o.username AS owner FROM shares s
shared = con.execute("""SELECT c.id, c.data, c.updated, c.npc, o.username AS owner FROM shares s
JOIN characters c ON c.id = s.character_id LEFT JOIN users o ON o.id = c.owner_id
WHERE s.user_id=? ORDER BY c.updated DESC""", (user["id"],)).fetchall()
return {"mine": [{"id": r["id"], "name": _name(r["data"]), "updated": r["updated"], "shared_with": shares.get(r["id"], [])}
for r in rows],
"shared": [{"id": r["id"], "name": _name(r["data"]), "updated": r["updated"], "owner": r["owner"] or "?"}
for r in shared]}
WHERE s.user_id=? AND c.npc=? ORDER BY c.updated DESC""", (user["id"], flag)).fetchall()
return {"mine": [_entry(r, shares=shares) for r in rows], "shared": [_entry(r, owner=r["owner"]) for r in shared]}
@api.post("/characters")
@@ -393,20 +416,39 @@ def create_character(ch: dict = Body(...), user: dict = Depends(current_user)):
if con.execute("SELECT COUNT(*) FROM characters WHERE owner_id=?", (user["id"],)).fetchone()[0] >= MAX_CHARACTERS:
raise HTTPException(409, "Character limit reached")
cid, now = uuid.uuid4().hex[:12], time.time()
con.execute("INSERT INTO characters (id, token, data, created, updated, owner_id) VALUES (?,?,?,?,?,?)",
(cid, "", json.dumps(ch), now, now, user["id"]))
con.execute("INSERT INTO characters (id, token, data, created, updated, owner_id, npc) VALUES (?,?,?,?,?,?,?)",
(cid, "", json.dumps(ch), now, now, user["id"], 1 if ch.get("npc") else 0))
return {"id": cid}
def _can_read_character(con, cid: str, user: dict) -> bool:
"""The owner, users the character is shared with, the DM of a lobby it plays in, and the DM of a lobby with a token of this NPC."""
row = con.execute("SELECT owner_id FROM characters WHERE id=?", (cid,)).fetchone()
return bool(row) and bool(row["owner_id"] == user["id"] or con.execute(
"SELECT 1 FROM shares WHERE character_id=? AND user_id=?", (cid, user["id"])).fetchone() or con.execute(
"""SELECT 1 FROM lobby_members m JOIN lobbies l ON l.id = m.lobby_id
WHERE m.character_id=? AND l.owner_id=?""", (cid, user["id"])).fetchone() or con.execute(
"""SELECT 1 FROM lobby_tokens t JOIN lobbies l ON l.id = t.lobby_id
WHERE t.npc_id=? AND l.owner_id=?""", (cid, user["id"])).fetchone())
@api.get("/characters/{cid}/portrait")
def character_portrait(cid: str, user: dict = Depends(current_user)):
with db() as con:
row = con.execute("SELECT data FROM characters WHERE id=?", (cid,)).fetchone()
allowed = _can_read_character(con, cid, user)
img = portrait_mod.image(json.loads(row["data"]).get("portrait")) if row and allowed else None
if not img:
raise HTTPException(404, "No portrait")
return Response(img, media_type="image/webp", headers={"Cache-Control": "private, max-age=86400"})
@api.get("/characters/{cid}")
def read_character(cid: str, user: dict = Depends(current_user)):
"""Readable by the owner, by users the character is shared with, and by the DM of a lobby it plays in."""
with db() as con:
row = con.execute("SELECT data, owner_id FROM characters WHERE id=?", (cid,)).fetchone()
allowed = bool(row) and (row["owner_id"] == user["id"] or con.execute(
"SELECT 1 FROM shares WHERE character_id=? AND user_id=?", (cid, user["id"])).fetchone() or con.execute(
"""SELECT 1 FROM lobby_members m JOIN lobbies l ON l.id = m.lobby_id
WHERE m.character_id=? AND l.owner_id=?""", (cid, user["id"])).fetchone())
allowed = _can_read_character(con, cid, user)
if not allowed:
raise HTTPException(404, "Character not found")
return json.loads(row["data"])
@@ -443,7 +485,7 @@ def update_character(cid: str, ch: dict = Body(...), user: dict = Depends(curren
check_size(ch)
with db() as con:
_owned(con, "characters", cid, user)
con.execute("UPDATE characters SET data=?, updated=? WHERE id=?", (json.dumps(ch), time.time(), cid))
con.execute("UPDATE characters SET data=?, updated=?, npc=? WHERE id=?", (json.dumps(ch), time.time(), 1 if ch.get("npc") else 0, cid))
lobbies = [r["lobby_id"] for r in con.execute("SELECT lobby_id FROM lobby_members WHERE character_id=?", (cid,))]
for lid in lobbies:
hub.notify(lid, "sheet", cid=cid)
@@ -485,7 +527,7 @@ def _new_code(con) -> str:
def _own_character(con, cid, user: dict):
row = con.execute("SELECT id FROM characters WHERE id=? AND owner_id=?", (cid, user["id"])).fetchone()
row = con.execute("SELECT id FROM characters WHERE id=? AND owner_id=? AND npc=0", (cid, user["id"])).fetchone()
if not row:
raise HTTPException(422, "Pick one of your own characters")
@@ -810,6 +852,14 @@ def roll_check(lid: str, body: dict = Body(...), user: dict = Depends(current_us
lobby = _lobby_for(con, lid, user)
member = con.execute("""SELECT m.char_mod, c.data FROM lobby_members m LEFT JOIN characters c ON c.id = m.character_id
WHERE m.lobby_id=? AND m.user_id=?""", (lid, user["id"])).fetchone()
npc_name = ""
if body.get("npc_token"): # the DM rolls for an NPC on the table
_dm_only(lobby, user)
t = con.execute("""SELECT t.name, c.data FROM lobby_tokens t JOIN characters c ON c.id = t.npc_id
WHERE t.id=? AND t.lobby_id=?""", (str(body["npc_token"]), lid)).fetchone()
if not t:
raise HTTPException(404, "This token has no NPC sheet")
member, npc_name = {"char_mod": 0, "data": t["data"]}, t["name"]
if not member or not member["data"]:
raise HTTPException(422, "Only players with a character in this lobby can roll checks")
c = compute(json.loads(member["data"]), rules)
@@ -819,12 +869,12 @@ def roll_check(lid: str, body: dict = Body(...), user: dict = Depends(current_us
raise HTTPException(404, "Unknown skill")
if not s["usable"]:
raise HTTPException(422, "This skill is still locked")
what, base = s["name"], s["total"]
what, base = (f"{npc_name}: " if npc_name else "") + s["name"], s["total"]
elif kind == "attr":
a = next((x for x in rules["attributes"] if x["id"] == key), None)
if not a:
raise HTTPException(404, "Unknown attribute")
what, base = a["name"], c["attrs"][key]
what, base = (f"{npc_name}: " if npc_name else "") + a["name"], c["attrs"][key]
else:
raise HTTPException(422, "Roll a skill or an attribute")
manual = _manual_results(None if body.get("manual") in (None, "") else [body.get("manual")], 100, 1)
@@ -975,6 +1025,43 @@ def admin_user(user: dict = Depends(current_user)) -> dict:
return user
@api.get("/admin/users")
def admin_users(user: dict = Depends(admin_user)):
names = {n.strip().lower() for n in os.environ.get("AW_ADMINS", "").split(",") if n.strip()}
with db() as con:
ids = admin_ids(con)
rows = con.execute("SELECT id, username, created FROM users ORDER BY username COLLATE NOCASE").fetchall()
return [{"username": r["username"], "admin": r["id"] in ids, "fixed": r["username"].lower() in names, "me": r["id"] == user["id"],
"created": r["created"]} for r in rows]
@api.put("/admin/users/{username}")
def set_admin(username: str, body: dict = Body(...), user: dict = Depends(admin_user)):
"""Make a user an admin or take it away again. Names listed in AW_ADMINS can only be changed in the server settings."""
want = bool(body.get("admin"))
names = {n.strip().lower() for n in os.environ.get("AW_ADMINS", "").split(",") if n.strip()}
with db() as con:
target = con.execute("SELECT id, username FROM users WHERE username=?", (username,)).fetchone()
if not target:
raise HTTPException(404, "No user with this name")
ids = admin_ids(con)
if want:
if not any(r["is_admin"] for r in con.execute("SELECT is_admin FROM users")) and not names: # the implicit first-account admin becomes explicit
con.execute("UPDATE users SET is_admin=1 WHERE id IN (%s)" % ",".join(str(i) for i in ids))
con.execute("UPDATE users SET is_admin=1 WHERE id=?", (target["id"],))
else:
if target["username"].lower() in names:
raise HTTPException(409, "This admin is set in AW_ADMINS on the server and can only be removed there")
if ids - {target["id"]} == set():
raise HTTPException(409, "There must be at least one admin")
if target["id"] not in ids:
return {"ok": True}
if not any(r["is_admin"] for r in con.execute("SELECT is_admin FROM users")): # implicit first account: make it explicit first
con.execute("UPDATE users SET is_admin=1 WHERE id IN (%s)" % ",".join(str(i) for i in ids))
con.execute("UPDATE users SET is_admin=0 WHERE id=?", (target["id"],))
return {"ok": True}
@api.get("/admin/rules")
def editor_state(user: dict = Depends(admin_user)):
"""The editable sections as they are now, and which of them differ from the shipped rules.json."""
@@ -1036,7 +1123,7 @@ def editor_reset(section: str, confirm: bool = False, user: dict = Depends(admin
return {"ok": True}
maps_mod.install(api, db, current_user, _lobby_for, _dm_only, hub, _event, Path(DB_PATH).parent / "maps")
maps_mod.install(api, db, current_user, _lobby_for, _dm_only, hub, _event, Path(DB_PATH).parent / "maps", _can_read_character)
purge_sounds = sounds_mod.install(api, db, current_user, _lobby_for, _dm_only, hub, Path(DB_PATH).parent / "sounds")