Lobby: 3-column layout with player area, skill checks, difficulty, colours, own dice, mystery dice
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01HJhXuLD2Z6hv3Wqeq9UL9K
This commit is contained in:
+196
-25
@@ -12,6 +12,7 @@ from fastapi.responses import FileResponse, JSONResponse
|
||||
from fastapi.staticfiles import StaticFiles
|
||||
|
||||
from . import auth
|
||||
from .derive import compute
|
||||
from .export import to_docx, to_pdf
|
||||
|
||||
BASE = Path(__file__).resolve().parent.parent
|
||||
@@ -26,6 +27,11 @@ MAX_LOBBIES = 20 # per DM
|
||||
MAX_MEMBERS = 20 # per lobby
|
||||
MAX_MESSAGE = 500 # characters per chat message
|
||||
KEEP_MESSAGES = 1000 # per lobby, older ones are dropped
|
||||
PALETTE = [ # player colours (dice, cards, chat names); every player in a lobby has a different one
|
||||
("red", "#e0524d"), ("orange", "#ea8a3b"), ("yellow", "#e6c52e"), ("green", "#6fc24b"), ("teal", "#2fb8a6"),
|
||||
("blue", "#4f93e6"), ("purple", "#9d6fda"), ("pink", "#e870b1"), ("white", "#e9e6da"), ("brown", "#a8764f")]
|
||||
PALETTE_HEX = dict(PALETTE)
|
||||
DM_COLOR = "#d4af37"
|
||||
CODE_ALPHABET = "ABCDEFGHJKLMNPQRSTUVWXYZ23456789" # no 0/O/1/I
|
||||
|
||||
# No public API docs: everything about the API stays behind the login
|
||||
@@ -62,6 +68,11 @@ def db() -> sqlite3.Connection:
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT, lobby_id TEXT NOT NULL, user_id INTEGER, to_user_id INTEGER,
|
||||
kind TEXT NOT NULL, text TEXT NOT NULL, created REAL NOT NULL)""")
|
||||
con.execute("CREATE INDEX IF NOT EXISTS lobby_messages_lobby ON lobby_messages (lobby_id, id)")
|
||||
for table, col, ddl in (("lobbies", "diff_mod", "INTEGER NOT NULL DEFAULT 0"), ("lobbies", "diff_label", "TEXT NOT NULL DEFAULT 'Challenging'"),
|
||||
("lobbies", "diff_hidden", "INTEGER NOT NULL DEFAULT 0"), ("lobby_members", "color", "TEXT"),
|
||||
("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}")
|
||||
for table in ("characters", "presets"): # older databases: add the owner column
|
||||
if "owner_id" not in [r["name"] for r in con.execute(f"PRAGMA table_info({table})")]:
|
||||
con.execute(f"ALTER TABLE {table} ADD COLUMN owner_id INTEGER")
|
||||
@@ -426,6 +437,32 @@ def _own_character(con, cid, user: dict):
|
||||
raise HTTPException(422, "Pick one of your own characters")
|
||||
|
||||
|
||||
def _free_color(con, lid: str):
|
||||
used = {r["color"] for r in con.execute("SELECT color FROM lobby_members WHERE lobby_id=?", (lid,))}
|
||||
return next((cid for cid, _ in PALETTE if cid not in used), PALETTE[0][0])
|
||||
|
||||
|
||||
def _ensure_colors(con, lid: str):
|
||||
"""Members that joined before colours existed get the next free one."""
|
||||
for r in con.execute("SELECT user_id FROM lobby_members WHERE lobby_id=? AND color IS NULL ORDER BY joined", (lid,)).fetchall():
|
||||
con.execute("UPDATE lobby_members SET color=? WHERE lobby_id=? AND user_id=?", (_free_color(con, lid), lid, r["user_id"]))
|
||||
|
||||
|
||||
def _levels(rules: dict) -> list:
|
||||
"""Difficulty levels from the rules text (section 'difficulty', first table): [{name, mod}]."""
|
||||
out = []
|
||||
for sec in rules.get("sections", []):
|
||||
if sec.get("id") != "difficulty":
|
||||
continue
|
||||
for block in sec.get("blocks", []):
|
||||
for row in (block.get("table") or {}).get("rows", []):
|
||||
try:
|
||||
out.append({"name": row[0], "mod": int(row[1].replace("\u2212", "-").replace("\u00b1", "").replace("+", "").strip() or 0)})
|
||||
except (ValueError, IndexError):
|
||||
pass
|
||||
return out
|
||||
|
||||
|
||||
def _event(con, lid: str, text: str, uid=None):
|
||||
"""Writes a line into the lobby's event log (shown in the chat feed to everybody)."""
|
||||
con.execute("INSERT INTO lobby_messages (lobby_id, user_id, kind, text, created) VALUES (?,?, 'event', ?, ?)",
|
||||
@@ -494,9 +531,9 @@ def join_lobby(request: Request, body: dict = Body(...), user: dict = Depends(cu
|
||||
known = con.execute("SELECT 1 FROM lobby_members WHERE lobby_id=? AND user_id=?", (lobby["id"], user["id"])).fetchone()
|
||||
if not known and con.execute("SELECT COUNT(*) FROM lobby_members WHERE lobby_id=?", (lobby["id"],)).fetchone()[0] >= MAX_MEMBERS:
|
||||
raise HTTPException(409, "This lobby is full")
|
||||
con.execute("""INSERT INTO lobby_members (lobby_id, user_id, character_id, joined) VALUES (?,?,?,?)
|
||||
con.execute("""INSERT INTO lobby_members (lobby_id, user_id, character_id, joined, color) VALUES (?,?,?,?,?)
|
||||
ON CONFLICT(lobby_id, user_id) DO UPDATE SET character_id=excluded.character_id""",
|
||||
(lobby["id"], user["id"], body["character_id"], time.time()))
|
||||
(lobby["id"], user["id"], body["character_id"], time.time(), _free_color(con, lobby["id"])))
|
||||
_event(con, lobby["id"], f"{user['username']} {'switched to' if known else 'joined as'} {_char_name(con, body['character_id'])}", user["id"])
|
||||
return {"id": lobby["id"]}
|
||||
|
||||
@@ -507,7 +544,8 @@ def read_lobby(lid: str, user: dict = Depends(current_user)):
|
||||
with db() as con:
|
||||
lobby = _lobby_for(con, lid, user)
|
||||
is_dm = lobby["owner_id"] == user["id"]
|
||||
rows = con.execute("""SELECT m.user_id, m.character_id, u.username, c.data FROM lobby_members m
|
||||
_ensure_colors(con, lid)
|
||||
rows = con.execute("""SELECT m.user_id, m.character_id, m.color, m.char_mod, u.username, c.data FROM lobby_members m
|
||||
JOIN users u ON u.id = m.user_id LEFT JOIN characters c ON c.id = m.character_id
|
||||
WHERE m.lobby_id=? ORDER BY m.joined""", (lid,)).fetchall()
|
||||
members = []
|
||||
@@ -515,10 +553,50 @@ def read_lobby(lid: str, user: dict = Depends(current_user)):
|
||||
mine = r["user_id"] == user["id"]
|
||||
card = _card(r["data"], rules) if r["data"] else None
|
||||
# The id is only given out where the viewer may open the full character (the DM sees all, a player only their own)
|
||||
members.append({"username": r["username"], "me": mine, "character": card,
|
||||
members.append({"username": r["username"], "me": mine, "character": card, "color": r["color"],
|
||||
"mod": r["char_mod"] if (is_dm or mine) else None,
|
||||
"character_id": r["character_id"] if (is_dm or mine) and card else None})
|
||||
hidden = bool(lobby["diff_hidden"])
|
||||
diff = {"label": lobby["diff_label"], "mod": lobby["diff_mod"], "hidden": hidden}
|
||||
if hidden and not is_dm:
|
||||
diff = {"label": "", "mod": None, "hidden": True}
|
||||
return {"id": lobby["id"], "name": lobby["name"], "dm": lobby["dm"], "is_dm": is_dm,
|
||||
"code": lobby["code"] if is_dm else None, "members": members}
|
||||
"code": lobby["code"] if is_dm else None, "members": members, "difficulty": diff,
|
||||
"levels": _levels(rules), "palette": [{"id": i, "hex": h} for i, h in PALETTE], "dm_color": DM_COLOR}
|
||||
|
||||
|
||||
@api.put("/lobbies/{lid}/difficulty")
|
||||
def set_difficulty(lid: str, body: dict = Body(...), user: dict = Depends(current_user)):
|
||||
"""The DM's current difficulty: a level from the rules table or a custom modifier. It applies to every check."""
|
||||
try:
|
||||
mod = int(body.get("mod", 0))
|
||||
except (TypeError, ValueError):
|
||||
raise HTTPException(422, "The modifier must be a whole number")
|
||||
if not -100 <= mod <= 100:
|
||||
raise HTTPException(422, "The modifier must be between -100 and 100")
|
||||
label = (str(body.get("label") or "").strip() or "Custom")[:24]
|
||||
with db() as con:
|
||||
_dm_only(_lobby_for(con, lid, user), user)
|
||||
con.execute("UPDATE lobbies SET diff_mod=?, diff_label=?, diff_hidden=? WHERE id=?", (mod, label, 1 if body.get("hidden") else 0, lid))
|
||||
return {"ok": True}
|
||||
|
||||
|
||||
@api.put("/lobbies/{lid}/members/{username}/mod")
|
||||
def set_member_mod(lid: str, username: str, body: dict = Body(...), user: dict = Depends(current_user)):
|
||||
"""A personal bonus or malus of one character (talent, condition ...), added on top of the lobby difficulty."""
|
||||
try:
|
||||
mod = int(body.get("mod", 0))
|
||||
except (TypeError, ValueError):
|
||||
raise HTTPException(422, "The modifier must be a whole number")
|
||||
if not -100 <= mod <= 100:
|
||||
raise HTTPException(422, "The modifier must be between -100 and 100")
|
||||
with db() as con:
|
||||
_dm_only(_lobby_for(con, lid, user), user)
|
||||
cur = con.execute("UPDATE lobby_members SET char_mod=? WHERE lobby_id=? AND user_id=(SELECT id FROM users WHERE username=?)",
|
||||
(mod, lid, username))
|
||||
if not cur.rowcount:
|
||||
raise HTTPException(404, "No such player in this lobby")
|
||||
return {"ok": True}
|
||||
|
||||
|
||||
def _recipient(con, lobby, user: dict, to):
|
||||
@@ -536,12 +614,16 @@ def _recipient(con, lobby, user: dict, to):
|
||||
|
||||
@api.get("/lobbies/{lid}/messages")
|
||||
def read_messages(lid: str, after: int = 0, user: dict = Depends(current_user)):
|
||||
"""Chat and event log. Whispers are only returned to the sender and the recipient (not even the DM sees them).
|
||||
"""Chat, event log, dice rolls and checks. Whispers are only returned to the sender and the recipient (not even the DM
|
||||
sees them). Mystery rolls of the DM come without results, hidden-difficulty checks without the numbers behind them.
|
||||
after=0 returns the latest 100 entries, otherwise everything newer than that id (the client polls)."""
|
||||
with db() as con:
|
||||
_lobby_for(con, lid, user)
|
||||
q = """SELECT m.id, m.kind, m.text, m.created, m.user_id, m.to_user_id, f.username AS sender, t.username AS target
|
||||
lobby = _lobby_for(con, lid, user)
|
||||
is_dm = lobby["owner_id"] == user["id"]
|
||||
q = """SELECT m.id, m.kind, m.text, m.created, m.user_id, m.to_user_id, f.username AS sender, t.username AS target,
|
||||
lm.color AS color
|
||||
FROM lobby_messages m LEFT JOIN users f ON f.id = m.user_id LEFT JOIN users t ON t.id = m.to_user_id
|
||||
LEFT JOIN lobby_members lm ON lm.lobby_id = m.lobby_id AND lm.user_id = m.user_id
|
||||
WHERE m.lobby_id=? AND (m.to_user_id IS NULL OR m.to_user_id=? OR m.user_id=?)"""
|
||||
if after > 0:
|
||||
rows = con.execute(q + " AND m.id>? ORDER BY m.id LIMIT 200", (lid, user["id"], user["id"], after)).fetchall()
|
||||
@@ -549,10 +631,22 @@ def read_messages(lid: str, after: int = 0, user: dict = Depends(current_user)):
|
||||
rows = list(reversed(con.execute(q + " ORDER BY m.id DESC LIMIT 100", (lid, user["id"], user["id"])).fetchall()))
|
||||
out = []
|
||||
for r in rows:
|
||||
mine = r["user_id"] == user["id"]
|
||||
item = {"id": r["id"], "kind": r["kind"], "text": r["text"], "time": r["created"], "from": r["sender"] or "",
|
||||
"to": r["target"] or "", "mine": r["user_id"] == user["id"] and r["kind"] != "event"}
|
||||
"to": r["target"] or "", "mine": mine and r["kind"] != "event",
|
||||
"color": DM_COLOR if r["user_id"] == lobby["owner_id"] else PALETTE_HEX.get(r["color"], "")}
|
||||
if r["kind"] == "roll":
|
||||
item["roll"], item["text"] = json.loads(r["text"]), ""
|
||||
roll = json.loads(r["text"])
|
||||
roll["n"] = len(roll["results"])
|
||||
if roll.get("mystery") and not mine:
|
||||
roll["results"] = None # players see that the DM rolled, not what
|
||||
item["roll"], item["text"] = roll, ""
|
||||
elif r["kind"] == "check":
|
||||
chk = json.loads(r["text"])
|
||||
if chk.get("hidden") and not is_dm:
|
||||
for k in ("base", "mod", "cmod", "diff", "target", "margin"):
|
||||
chk.pop(k, None)
|
||||
item["check"], item["text"] = chk, ""
|
||||
out.append(item)
|
||||
return out
|
||||
|
||||
@@ -580,10 +674,32 @@ DICE = (4, 6, 8, 10, 12, 20, 100)
|
||||
MAX_DICE = 10
|
||||
|
||||
|
||||
def _manual_results(raw, sides: int, count: int):
|
||||
"""Results typed in by a player who rolled real dice. None when nothing was typed."""
|
||||
if raw in (None, [], ""):
|
||||
return None
|
||||
try:
|
||||
vals = [int(v) for v in raw]
|
||||
except (TypeError, ValueError):
|
||||
raise HTTPException(422, "Enter whole numbers")
|
||||
if len(vals) != count or any(not 1 <= v <= sides for v in vals):
|
||||
raise HTTPException(422, f"Enter {count} number{'s' if count > 1 else ''} from 1 to {sides}")
|
||||
return vals
|
||||
|
||||
|
||||
def _post(con, lid: str, uid: int, to_id, kind: str, payload: dict) -> int:
|
||||
cur = con.execute("INSERT INTO lobby_messages (lobby_id, user_id, to_user_id, kind, text, created) VALUES (?,?,?,?,?,?)",
|
||||
(lid, uid, to_id, kind, json.dumps(payload), time.time()))
|
||||
con.execute("""DELETE FROM lobby_messages WHERE lobby_id=? AND id <= COALESCE(
|
||||
(SELECT id FROM lobby_messages WHERE lobby_id=? ORDER BY id DESC LIMIT 1 OFFSET ?), 0)""", (lid, lid, KEEP_MESSAGES))
|
||||
return cur.lastrowid
|
||||
|
||||
|
||||
@api.post("/lobbies/{lid}/rolls")
|
||||
def roll_dice(lid: str, body: dict = Body(...), user: dict = Depends(current_user)):
|
||||
"""Rolls on the server (so nobody can cheat) and posts the result in the chat; with `to` it is a private roll.
|
||||
Rolls by the DM are only visible to the DM, or to the person picked with `to`."""
|
||||
Players with real dice can send their own results in `manual` (marked as such in the chat).
|
||||
The DM picks `mode`: hidden (only the DM sees it), mystery (players see that dice were rolled, not the result) or public."""
|
||||
try:
|
||||
sides, count = int(body.get("sides")), int(body.get("count", 1))
|
||||
except (TypeError, ValueError):
|
||||
@@ -592,17 +708,62 @@ def roll_dice(lid: str, body: dict = Body(...), user: dict = Depends(current_use
|
||||
raise HTTPException(422, f"Roll 1 to {MAX_DICE} dice of: " + ", ".join(f"d{d}" for d in DICE))
|
||||
throttle(f"chat:{user['id']}", 30, 60)
|
||||
limiter.fail(f"chat:{user['id']}")
|
||||
results = [secrets.randbelow(sides) + 1 for _ in range(count)]
|
||||
manual = _manual_results(body.get("manual"), sides, count)
|
||||
results = manual or [secrets.randbelow(sides) + 1 for _ in range(count)]
|
||||
with db() as con:
|
||||
lobby = _lobby_for(con, lid, user)
|
||||
is_dm = lobby["owner_id"] == user["id"]
|
||||
to_id = _recipient(con, lobby, user, body.get("to"))
|
||||
if to_id is None and lobby["owner_id"] == user["id"]:
|
||||
to_id = user["id"] # the DM's rolls are hidden from the players (unless the DM picks someone to show them to)
|
||||
cur = con.execute("INSERT INTO lobby_messages (lobby_id, user_id, to_user_id, kind, text, created) VALUES (?,?,?,?,?,?)",
|
||||
(lid, user["id"], to_id, "roll", json.dumps({"sides": sides, "results": results}), time.time()))
|
||||
con.execute("""DELETE FROM lobby_messages WHERE lobby_id=? AND id <= COALESCE(
|
||||
(SELECT id FROM lobby_messages WHERE lobby_id=? ORDER BY id DESC LIMIT 1 OFFSET ?), 0)""", (lid, lid, KEEP_MESSAGES))
|
||||
return {"id": cur.lastrowid, "sides": sides, "results": results}
|
||||
mystery = False
|
||||
if is_dm and to_id is None:
|
||||
mode = body.get("mode") or "hidden"
|
||||
if mode not in ("hidden", "mystery", "public"):
|
||||
raise HTTPException(422, "Unknown roll mode")
|
||||
if mode == "hidden":
|
||||
to_id = user["id"] # only the DM sees it
|
||||
mystery = mode == "mystery"
|
||||
mid = _post(con, lid, user["id"], to_id, "roll", {"sides": sides, "results": results, "mystery": mystery, "manual": bool(manual)})
|
||||
return {"id": mid, "sides": sides, "results": results}
|
||||
|
||||
|
||||
@api.post("/lobbies/{lid}/checks")
|
||||
def roll_check(lid: str, body: dict = Body(...), user: dict = Depends(current_user)):
|
||||
"""A d100 roll-under check for a skill or attribute of the player's current lobby character.
|
||||
Target = attribute + skill + the DM's difficulty + the character's personal modifier. 01-05 always succeeds, 96-100 always fails."""
|
||||
rules = load_rules()
|
||||
kind, key = str(body.get("kind", "")), str(body.get("id", ""))
|
||||
throttle(f"chat:{user['id']}", 30, 60)
|
||||
limiter.fail(f"chat:{user['id']}")
|
||||
with db() as con:
|
||||
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()
|
||||
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)
|
||||
if kind == "skill":
|
||||
s = next((x for x in c["skills"] if x["id"] == key), None)
|
||||
if not s:
|
||||
raise HTTPException(404, "Unknown skill")
|
||||
if not s["usable"]:
|
||||
raise HTTPException(422, "This skill is still locked")
|
||||
what, base = 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]
|
||||
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)
|
||||
roll = manual[0] if manual else secrets.randbelow(100) + 1
|
||||
target = base + lobby["diff_mod"] + member["char_mod"]
|
||||
outcome = "crit_success" if roll <= 5 else "crit_failure" if roll >= 96 else "success" if roll <= target else "failure"
|
||||
payload = {"kind": kind, "what": what, "base": base, "mod": lobby["diff_mod"], "cmod": member["char_mod"], "diff": lobby["diff_label"],
|
||||
"target": target, "roll": roll, "margin": target - roll, "outcome": outcome, "hidden": bool(lobby["diff_hidden"]),
|
||||
"manual": bool(manual)}
|
||||
mid = _post(con, lid, user["id"], None, "check", payload)
|
||||
return {"id": mid, "roll": roll, "outcome": outcome}
|
||||
|
||||
|
||||
@api.put("/lobbies/{lid}")
|
||||
@@ -639,15 +800,25 @@ def new_lobby_code(lid: str, user: dict = Depends(current_user)):
|
||||
|
||||
|
||||
@api.put("/lobbies/{lid}/me")
|
||||
def switch_lobby_character(lid: str, body: dict = Body(...), user: dict = Depends(current_user)):
|
||||
def update_my_membership(lid: str, body: dict = Body(...), user: dict = Depends(current_user)):
|
||||
"""Switch your character and/or pick your colour (no two players share one)."""
|
||||
with db() as con:
|
||||
_lobby_for(con, lid, user)
|
||||
_own_character(con, body.get("character_id"), user)
|
||||
cur = con.execute("UPDATE lobby_members SET character_id=? WHERE lobby_id=? AND user_id=?",
|
||||
(body["character_id"], lid, user["id"]))
|
||||
if not cur.rowcount:
|
||||
if not con.execute("SELECT 1 FROM lobby_members WHERE lobby_id=? AND user_id=?", (lid, user["id"])).fetchone():
|
||||
raise HTTPException(422, "You are the DM of this lobby")
|
||||
_event(con, lid, f"{user['username']} switched to {_char_name(con, body['character_id'])}", user["id"])
|
||||
if "character_id" not in body and "color" not in body:
|
||||
raise HTTPException(422, "Nothing to change")
|
||||
if "color" in body:
|
||||
color = body["color"]
|
||||
if color not in PALETTE_HEX:
|
||||
raise HTTPException(422, "Unknown colour")
|
||||
if con.execute("SELECT 1 FROM lobby_members WHERE lobby_id=? AND color=? AND user_id!=?", (lid, color, user["id"])).fetchone():
|
||||
raise HTTPException(409, "Another player already uses this colour")
|
||||
con.execute("UPDATE lobby_members SET color=? WHERE lobby_id=? AND user_id=?", (color, lid, user["id"]))
|
||||
if "character_id" in body:
|
||||
_own_character(con, body.get("character_id"), user)
|
||||
con.execute("UPDATE lobby_members SET character_id=? WHERE lobby_id=? AND user_id=?", (body["character_id"], lid, user["id"]))
|
||||
_event(con, lid, f"{user['username']} switched to {_char_name(con, body['character_id'])}", user["id"])
|
||||
return {"ok": True}
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user