Rules editor: admins can edit items, skills, professions (and categories/groups) in the app
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01HJhXuLD2Z6hv3Wqeq9UL9K
This commit is contained in:
+107
-4
@@ -13,6 +13,7 @@ from fastapi.staticfiles import StaticFiles
|
||||
|
||||
from . import auth
|
||||
from .derive import compute
|
||||
from . import rules_edit
|
||||
from .export import to_docx, to_pdf
|
||||
|
||||
BASE = Path(__file__).resolve().parent.parent
|
||||
@@ -41,11 +42,30 @@ TRUST_PROXY = os.environ.get("AW_TRUST_PROXY", "") == "1" # behind a reverse pr
|
||||
|
||||
|
||||
|
||||
def load_rules() -> dict:
|
||||
# Re-read on every request so rules.json can be edited without a restart.
|
||||
# Changes made in the rules editor live next to the database (writable), the shipped rules.json stays untouched.
|
||||
OVERRIDES_PATH = Path(DB_PATH).parent / "rules_overrides.json"
|
||||
|
||||
|
||||
def base_rules() -> dict:
|
||||
return json.loads(RULES_PATH.read_text(encoding="utf-8"))
|
||||
|
||||
|
||||
def read_overrides() -> dict:
|
||||
try:
|
||||
return json.loads(OVERRIDES_PATH.read_text(encoding="utf-8"))
|
||||
except (OSError, ValueError):
|
||||
return {}
|
||||
|
||||
|
||||
def load_rules() -> dict:
|
||||
# Re-read on every request so rules.json can be edited without a restart; editor changes replace whole sections.
|
||||
r = base_rules()
|
||||
for sec, items in read_overrides().items():
|
||||
if sec in rules_edit.SECTIONS:
|
||||
r[sec] = items
|
||||
return r
|
||||
|
||||
|
||||
def db() -> sqlite3.Connection:
|
||||
Path(DB_PATH).parent.mkdir(parents=True, exist_ok=True)
|
||||
con = sqlite3.connect(DB_PATH)
|
||||
@@ -154,6 +174,16 @@ 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."""
|
||||
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:
|
||||
first = con.execute("SELECT id FROM users ORDER BY id LIMIT 1").fetchone()
|
||||
return bool(first) and first["id"] == user["id"]
|
||||
|
||||
|
||||
def current_user(request: Request) -> dict:
|
||||
with db() as con:
|
||||
user = auth.user_for_token(con, request.cookies.get(auth.SESSION_COOKIE))
|
||||
@@ -187,6 +217,8 @@ def check_password_rules(pw: str):
|
||||
def me(request: Request):
|
||||
with db() as con:
|
||||
user = auth.user_for_token(con, request.cookies.get(auth.SESSION_COOKIE))
|
||||
if user:
|
||||
user = {**user, "admin": is_admin(user)}
|
||||
return {"user": user, "registration_open": bool(auth.registration_code())}
|
||||
|
||||
|
||||
@@ -213,7 +245,8 @@ def register(request: Request, response: Response, body: dict = Body(...)):
|
||||
(username, auth.hash_password(password), time.time()))
|
||||
token = auth.create_session(con, cur.lastrowid)
|
||||
set_cookie(request, response, token)
|
||||
return {"user": {"id": cur.lastrowid, "username": username}}
|
||||
u = {"id": cur.lastrowid, "username": username}
|
||||
return {"user": {**u, "admin": is_admin(u)}}
|
||||
|
||||
|
||||
@app.post("/api/auth/login")
|
||||
@@ -235,7 +268,8 @@ def login(request: Request, response: Response, body: dict = Body(...)):
|
||||
token = auth.create_session(con, row["id"])
|
||||
limiter.reset(keys[1])
|
||||
set_cookie(request, response, token)
|
||||
return {"user": {"id": row["id"], "username": row["username"]}}
|
||||
u = {"id": row["id"], "username": row["username"]}
|
||||
return {"user": {**u, "admin": is_admin(u)}}
|
||||
|
||||
|
||||
@app.post("/api/auth/logout")
|
||||
@@ -883,6 +917,75 @@ def export(fmt: str, ch: dict = Body(...), user: dict = Depends(current_user)):
|
||||
headers={"Content-Disposition": f'attachment; filename="{_filename(ch, fmt)}"'})
|
||||
|
||||
|
||||
# ---- rules editor (admins only)
|
||||
|
||||
def admin_user(user: dict = Depends(current_user)) -> dict:
|
||||
if not is_admin(user):
|
||||
raise HTTPException(403, "Only admins can edit the rules")
|
||||
return user
|
||||
|
||||
|
||||
@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."""
|
||||
r = load_rules()
|
||||
return {"sections": {k: r[k] for k in rules_edit.SECTIONS}, "customized": [k for k in read_overrides() if k in rules_edit.SECTIONS]}
|
||||
|
||||
|
||||
def _all_characters(con) -> list:
|
||||
return [json.loads(x["data"]) for x in con.execute("SELECT data FROM characters")]
|
||||
|
||||
|
||||
def _save_overrides(data: dict):
|
||||
OVERRIDES_PATH.parent.mkdir(parents=True, exist_ok=True)
|
||||
if OVERRIDES_PATH.exists():
|
||||
OVERRIDES_PATH.with_suffix(".json.bak").write_bytes(OVERRIDES_PATH.read_bytes())
|
||||
tmp = OVERRIDES_PATH.with_suffix(".json.tmp")
|
||||
tmp.write_text(json.dumps(data, ensure_ascii=False, indent=1), encoding="utf-8")
|
||||
os.replace(tmp, OVERRIDES_PATH)
|
||||
|
||||
|
||||
def _apply(section: str, new: list, confirm: bool):
|
||||
r = load_rules()
|
||||
old_ids = {x["id"] for x in r[section]}
|
||||
removed = old_ids - {x["id"] for x in new}
|
||||
bad = rules_edit.dangling(section, new, r)
|
||||
if bad:
|
||||
raise HTTPException(422, "Cannot save: " + "; ".join(bad[:5]) + (" ..." if len(bad) > 5 else ""))
|
||||
with db() as con:
|
||||
used = rules_edit.usage(_all_characters(con), section, removed)
|
||||
if used and not confirm:
|
||||
raise HTTPException(409, f"{used} saved character{'s' if used != 1 else ''} use{'' if used != 1 else 's'} entries you are removing. "
|
||||
"Their data stays in the characters but the removed entries disappear from the sheets.")
|
||||
return used
|
||||
|
||||
|
||||
@api.put("/admin/rules/{section}")
|
||||
def editor_save(section: str, body: dict = Body(...), user: dict = Depends(admin_user)):
|
||||
try:
|
||||
new = rules_edit.clean(section, body.get("items"), load_rules())
|
||||
except rules_edit.RuleError as e:
|
||||
raise HTTPException(422, str(e))
|
||||
_apply(section, new, bool(body.get("confirm")))
|
||||
data = read_overrides()
|
||||
data[section] = new
|
||||
_save_overrides(data)
|
||||
return {"ok": True, "count": len(new)}
|
||||
|
||||
|
||||
@api.delete("/admin/rules/{section}")
|
||||
def editor_reset(section: str, confirm: bool = False, user: dict = Depends(admin_user)):
|
||||
"""Back to the shipped rules.json for this section."""
|
||||
if section not in rules_edit.SECTIONS:
|
||||
raise HTTPException(404, "Unknown section")
|
||||
data = read_overrides()
|
||||
if section in data:
|
||||
_apply(section, base_rules()[section], confirm)
|
||||
del data[section]
|
||||
_save_overrides(data)
|
||||
return {"ok": True}
|
||||
|
||||
|
||||
app.include_router(api)
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,125 @@
|
||||
"""Validation for the rules editor (items, skills, professions and their categories).
|
||||
Only whitelisted fields are kept and every reference (category, group, attribute, skill) has to exist."""
|
||||
import re
|
||||
|
||||
SECTIONS = ("equipment", "equipment_categories", "skills", "skill_groups", "professions")
|
||||
MAX_ENTRIES = 400
|
||||
ID_RE = re.compile(r"^[a-z0-9_]{1,40}$")
|
||||
|
||||
|
||||
class RuleError(ValueError):
|
||||
pass
|
||||
|
||||
|
||||
def _text(v, limit, label, required=False):
|
||||
v = str(v if v is not None else "").strip()
|
||||
if required and not v:
|
||||
raise RuleError(f"{label} must not be empty")
|
||||
if len(v) > limit:
|
||||
raise RuleError(f"{label} is too long (max {limit} characters)")
|
||||
return v
|
||||
|
||||
|
||||
def _num(v, lo, hi, label, integer=False):
|
||||
try:
|
||||
n = float(v)
|
||||
except (TypeError, ValueError):
|
||||
raise RuleError(f"{label} must be a number")
|
||||
if not lo <= n <= hi:
|
||||
raise RuleError(f"{label} must be between {lo} and {hi}")
|
||||
return int(n) if integer else round(n, 2)
|
||||
|
||||
|
||||
def _ids(v, allowed, label, lo=0, hi=50):
|
||||
if not isinstance(v, list) or not lo <= len(v) <= hi:
|
||||
raise RuleError(f"{label}: choose {lo} to {hi}" if hi < 50 else f"{label} must be a list")
|
||||
out = []
|
||||
for x in v:
|
||||
if x not in allowed:
|
||||
raise RuleError(f"{label}: unknown entry '{x}'")
|
||||
if x not in out:
|
||||
out.append(x)
|
||||
return out
|
||||
|
||||
|
||||
def clean(section: str, items, rules: dict) -> list:
|
||||
"""Returns the cleaned list for one section. `rules` is the effective rules (for the other sections' ids)."""
|
||||
if section not in SECTIONS:
|
||||
raise RuleError("Unknown section")
|
||||
if not isinstance(items, list) or len(items) > MAX_ENTRIES:
|
||||
raise RuleError(f"A list of at most {MAX_ENTRIES} entries is expected")
|
||||
attrs = {a["id"] for a in rules["attributes"]}
|
||||
out, seen = [], set()
|
||||
cats = {c["id"] for c in (items if section == "equipment_categories" else rules["equipment_categories"])}
|
||||
groups = {g["id"] for g in (items if section == "skill_groups" else rules["skill_groups"])}
|
||||
skills = {s["id"] for s in (items if section == "skills" else rules["skills"])}
|
||||
for it in items:
|
||||
if not isinstance(it, dict):
|
||||
raise RuleError("Every entry has to be an object")
|
||||
iid = str(it.get("id", ""))
|
||||
if not ID_RE.match(iid):
|
||||
raise RuleError(f"Invalid id '{iid}' (lowercase letters, digits and _ only, up to 40)")
|
||||
if iid in seen:
|
||||
raise RuleError(f"The id '{iid}' is used twice")
|
||||
seen.add(iid)
|
||||
name = _text(it.get("name"), 60, f"Name of '{iid}'", True)
|
||||
e = {"id": iid, "name": name}
|
||||
label = f"'{name}'"
|
||||
if section == "equipment":
|
||||
if it.get("cat") not in cats:
|
||||
raise RuleError(f"{label}: unknown category")
|
||||
e.update(cat=it["cat"], stat=_text(it.get("stat"), 40, f"{label} stat"), kg=_num(it.get("kg", 0), 0, 1000, f"{label} weight"),
|
||||
note=_text(it.get("note"), 200, f"{label} note"))
|
||||
if it.get("tier"):
|
||||
e["tier"] = _text(it["tier"], 20, f"{label} tier")
|
||||
elif section == "equipment_categories":
|
||||
e["stat"] = _text(it.get("stat"), 40, f"{label} stat column")
|
||||
elif section == "skills":
|
||||
if it.get("group") not in groups:
|
||||
raise RuleError(f"{label}: unknown group")
|
||||
if it.get("type") not in ("basic", "advanced"):
|
||||
raise RuleError(f"{label}: type must be basic or advanced")
|
||||
a = _ids(it.get("attrs"), attrs, f"{label} attributes", 1, 3)
|
||||
e.update(group=it["group"], attrs=a, type=it["type"], desc=_text(it.get("desc"), 300, f"{label} description"))
|
||||
elif section == "professions":
|
||||
e.update(desc=_text(it.get("desc"), 500, f"{label} description"), bonus_attrs=_ids(it.get("bonus_attrs"), attrs, f"{label} bonus attributes", 0, 7),
|
||||
proficiencies=_ids(it.get("proficiencies"), skills, f"{label} proficiencies", 0, 12),
|
||||
picks=_num(it.get("picks", 0), 0, 10, f"{label} free picks", True), note=_text(it.get("note"), 200, f"{label} note"))
|
||||
if not e["note"]:
|
||||
del e["note"]
|
||||
if it.get("placeholder"):
|
||||
e["placeholder"] = True
|
||||
out.append(e)
|
||||
return out
|
||||
|
||||
|
||||
def dangling(section: str, new: list, rules: dict) -> list:
|
||||
"""Things in the other sections that would point to an entry that no longer exists."""
|
||||
ids = {x["id"] for x in new}
|
||||
msgs = []
|
||||
if section == "equipment_categories":
|
||||
msgs += [f"Item '{e['name']}' uses the removed category" for e in rules["equipment"] if e["cat"] not in ids]
|
||||
if section == "skill_groups":
|
||||
msgs += [f"Skill '{s['name']}' uses the removed group" for s in rules["skills"] if s["group"] not in ids]
|
||||
if section == "skills":
|
||||
for p in rules["professions"]:
|
||||
gone = [i for i in p["proficiencies"] if i not in ids]
|
||||
if gone:
|
||||
msgs.append(f"Profession '{p['name']}' grants proficiency in a removed skill ({', '.join(gone)})")
|
||||
return msgs
|
||||
|
||||
|
||||
def character_refs(ch: dict) -> dict:
|
||||
"""Ids a saved character points to, per section."""
|
||||
kit = ch.get("kit") or {}
|
||||
items = {i for i in [kit.get("weapon"), kit.get("armor"), kit.get("tool"), *(kit.get("resources") or []), *(kit.get("extras") or {})] if i}
|
||||
sk = set((ch.get("skill_points") or {}).keys()) | set((ch.get("proficiencies") or {}).keys()) | set((ch.get("proficiency_picks") or {}).keys()) \
|
||||
| set(ch.get("unlocked") or []) | set(ch.get("profession_picks") or [])
|
||||
return {"equipment": items, "skills": sk, "professions": {ch["profession"]} if ch.get("profession") else set()}
|
||||
|
||||
|
||||
def usage(chars: list, section: str, removed: set) -> int:
|
||||
"""How many saved characters use one of the removed ids."""
|
||||
if section not in ("equipment", "skills", "professions") or not removed:
|
||||
return 0
|
||||
return sum(1 for ch in chars if character_refs(ch)[section] & removed)
|
||||
Reference in New Issue
Block a user