First slice: classes, combat, questing, Deadmines
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
commit
71dd99795a
17 changed files with 4522 additions and 0 deletions
411
app.py
Normal file
411
app.py
Normal file
|
|
@ -0,0 +1,411 @@
|
|||
#!/usr/bin/env python3
|
||||
"""Small, dependency-free server for World of Guildhall."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import hashlib
|
||||
import json
|
||||
import logging
|
||||
import mimetypes
|
||||
import random
|
||||
import re
|
||||
import sqlite3
|
||||
import threading
|
||||
import time
|
||||
from contextlib import closing
|
||||
from datetime import date
|
||||
from http import HTTPStatus
|
||||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||||
from pathlib import Path
|
||||
from urllib.parse import unquote, urlparse
|
||||
|
||||
import combat
|
||||
import content
|
||||
import game
|
||||
|
||||
APP_DIR = Path(__file__).resolve().parent
|
||||
STATIC_DIR = APP_DIR / "static"
|
||||
MAX_BODY = 16 * 1024
|
||||
NAME = re.compile(r"^[A-Za-z][A-Za-z'-]{1,15}$")
|
||||
TICK_SECONDS = 0.25
|
||||
SAVE_SECONDS = 5
|
||||
|
||||
|
||||
def build_version() -> str:
|
||||
"""Changes whenever the code or client does; open tabs reload when it does."""
|
||||
digest = hashlib.sha256()
|
||||
for path in sorted([*APP_DIR.glob("*.py"), *STATIC_DIR.rglob("*")]):
|
||||
if path.is_file():
|
||||
digest.update(path.name.encode())
|
||||
digest.update(path.read_bytes())
|
||||
return digest.hexdigest()[:12]
|
||||
|
||||
|
||||
VERSION = build_version()
|
||||
|
||||
mimetypes.add_type("application/manifest+json", ".webmanifest")
|
||||
|
||||
|
||||
class Store:
|
||||
"""The world lives in memory and is written whole to SQLite every few seconds."""
|
||||
|
||||
def __init__(self, path: Path, dev: bool):
|
||||
self.path = path
|
||||
self.lock = threading.Lock()
|
||||
self.rng = random.Random()
|
||||
self.saved = 0.0
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
with closing(sqlite3.connect(path)) as db, db:
|
||||
db.execute("CREATE TABLE IF NOT EXISTS world (id INTEGER PRIMARY KEY CHECK (id = 1), data TEXT NOT NULL)")
|
||||
row = db.execute("SELECT data FROM world WHERE id = 1").fetchone()
|
||||
self.world = json.loads(row[0]) if row else game.new_world(time.time())
|
||||
self.world["config"]["dev"] = dev
|
||||
self.save()
|
||||
|
||||
def now(self) -> float:
|
||||
return time.time() + self.world["config"]["clock_offset"]
|
||||
|
||||
def save(self) -> None:
|
||||
data = json.dumps(self.world, separators=(",", ":"))
|
||||
with closing(sqlite3.connect(self.path)) as db, db:
|
||||
db.execute("INSERT INTO world (id, data) VALUES (1, ?) ON CONFLICT (id) DO UPDATE SET data = excluded.data",
|
||||
(data,))
|
||||
self.saved = time.time()
|
||||
|
||||
def backup(self, destination: Path, keep: int) -> Path:
|
||||
"""Write today's consistent snapshot and prune all but the newest `keep`."""
|
||||
destination.mkdir(parents=True, exist_ok=True)
|
||||
target = destination / f"wog-{date.today().isoformat()}.sqlite3"
|
||||
partial = target.with_suffix(".partial")
|
||||
with closing(sqlite3.connect(self.path)) as source, closing(sqlite3.connect(partial)) as copy:
|
||||
source.backup(copy)
|
||||
partial.replace(target)
|
||||
snapshots = sorted(destination.glob("wog-*.sqlite3"))
|
||||
for old in snapshots[:-keep] if keep > 0 else []:
|
||||
old.unlink()
|
||||
return target
|
||||
|
||||
def tick(self) -> None:
|
||||
with self.lock:
|
||||
game.process(self.world, self.now(), self.rng)
|
||||
if time.time() - self.saved > SAVE_SECONDS:
|
||||
self.save()
|
||||
|
||||
|
||||
def static_content() -> dict:
|
||||
referenced = set(content.ITEMS)
|
||||
for quest in content.QUESTS.values():
|
||||
referenced.update(quest.get("choice", []))
|
||||
return {
|
||||
"classes": content.CLASSES, "specs": content.SPECS, "abilities": content.ABILITIES, "unlock": content.UNLOCK,
|
||||
"places": content.PLACES, "quests": content.QUESTS, "dungeons": content.DUNGEONS,
|
||||
"mobs": {k: {"name": m["name"], "lv": m["lv"], "rare": m.get("rare", False), "boss": m.get("boss", False),
|
||||
"elite": m.get("elite", False)} for k, m in content.MOBS.items()},
|
||||
"enemy_abilities": content.ENEMY_ABILITIES, "slots": content.SLOTS, "slot_names": content.SLOT_NAMES,
|
||||
"xp": content.XP_TO_NEXT, "max_level": content.MAX_LEVEL,
|
||||
"items": {i: game.item(i) for i in sorted(referenced)},
|
||||
"auto_cd": combat.AUTO_CD, "watch": combat.WATCH, "gcd": combat.GCD,
|
||||
}
|
||||
|
||||
|
||||
CONTENT = json.dumps(static_content(), separators=(",", ":")).encode()
|
||||
|
||||
|
||||
def player_view(world: dict, player: dict, now: float) -> dict:
|
||||
st = game.stats(player)
|
||||
hp, mana = game.hp_now(player, now)
|
||||
view = {key: player[key] for key in ("id", "name", "cls", "level", "xp", "copper", "place", "activity",
|
||||
"quests", "done", "manual", "group", "dead_until", "next_pull")}
|
||||
view["hp"], view["mana"] = round(hp), round(mana, 3)
|
||||
view["stats"] = st
|
||||
view["xp_next"] = content.XP_TO_NEXT[player["level"]]
|
||||
view["known"] = game.known(player)
|
||||
view["equip"] = {slot: game.item(i) if i else None for slot, i in player["equip"].items()}
|
||||
bags = []
|
||||
for item_id, qty in sorted(player["bags"].items()):
|
||||
spec = game.item(item_id)
|
||||
if spec:
|
||||
bags.append({**spec, "qty": qty, "fit": game.can_wear(player, spec) if "slot" in spec else None})
|
||||
view["bags"] = bags
|
||||
view["available"] = game.available_quests(player)
|
||||
view["log"] = player["log"][-40:]
|
||||
return view
|
||||
|
||||
|
||||
def group_view(world: dict, group: dict | None, now: float) -> dict | None:
|
||||
if not group:
|
||||
return None
|
||||
members = []
|
||||
for uid in group["members"]:
|
||||
member = world["players"][uid]
|
||||
hp, mana = game.hp_now(member, now)
|
||||
members.append({"id": uid, "name": member["name"], "cls": member["cls"], "level": member["level"],
|
||||
"place": member["place"], "hp": round(hp), "max_hp": game.stats(member)["max_hp"],
|
||||
"mana": round(mana, 2), "inside": bool(member["activity"]
|
||||
and member["activity"]["kind"] == "dungeon"),
|
||||
"busy": game._busy(member)})
|
||||
dungeon = {k: v for k, v in group["dungeon"].items() if k != "last_fight"} if group["dungeon"] else None
|
||||
return {"id": group["id"], "leader": group["leader"], "dungeon": dungeon, "members": members}
|
||||
|
||||
|
||||
def roster(world: dict, now: float) -> list[dict]:
|
||||
people = []
|
||||
for player in world["players"].values():
|
||||
activity = player["activity"]
|
||||
people.append({"id": player["id"], "name": player["name"], "cls": player["cls"], "level": player["level"],
|
||||
"place": player["place"], "group": player["group"],
|
||||
"doing": activity["kind"] if activity else None,
|
||||
"hunt": activity.get("hunt") if activity else None,
|
||||
"online": now - player["last_active"] < 120})
|
||||
return sorted(people, key=lambda p: (-p["level"], p["name"]))
|
||||
|
||||
|
||||
class Handler(BaseHTTPRequestHandler):
|
||||
server_version = "WoG"
|
||||
store: Store
|
||||
dev = False
|
||||
user_header = "Remote-User"
|
||||
|
||||
def log_message(self, fmt: str, *args) -> None:
|
||||
logging.debug("%s - %s", self.address_string(), fmt % args)
|
||||
|
||||
# --- plumbing --------------------------------------------------------
|
||||
|
||||
def identity(self) -> str | None:
|
||||
if self.dev:
|
||||
cookie = self.headers.get("Cookie", "")
|
||||
match = re.search(r"(?:^|;\s*)wog_user=([^;]+)", cookie)
|
||||
return unquote(match.group(1)) if match else "fred"
|
||||
user = self.headers.get(self.user_header, "").strip()
|
||||
return user or None
|
||||
|
||||
def send_bytes(self, body: bytes, kind: str = "application/json", cache: str = "no-store") -> None:
|
||||
self.send_response(HTTPStatus.OK)
|
||||
self.send_header("Content-Type", kind)
|
||||
self.send_header("Content-Length", str(len(body)))
|
||||
self.send_header("Cache-Control", cache)
|
||||
self.end_headers()
|
||||
self.wfile.write(body)
|
||||
|
||||
def send_json(self, payload: object, status: HTTPStatus = HTTPStatus.OK) -> None:
|
||||
body = json.dumps(payload, separators=(",", ":")).encode()
|
||||
self.send_response(status)
|
||||
self.send_header("Content-Type", "application/json")
|
||||
self.send_header("Content-Length", str(len(body)))
|
||||
self.send_header("Cache-Control", "no-store")
|
||||
self.end_headers()
|
||||
self.wfile.write(body)
|
||||
|
||||
def fail(self, message: str, status: HTTPStatus = HTTPStatus.BAD_REQUEST) -> None:
|
||||
self.send_json({"error": message}, status)
|
||||
|
||||
def read_body(self) -> dict | None:
|
||||
length = int(self.headers.get("Content-Length") or 0)
|
||||
if length > MAX_BODY:
|
||||
return None
|
||||
try:
|
||||
data = json.loads(self.rfile.read(length) or b"{}")
|
||||
except json.JSONDecodeError:
|
||||
return None
|
||||
return data if isinstance(data, dict) else None
|
||||
|
||||
def send_static(self, path: str) -> None:
|
||||
name = "index.html" if path in ("", "/") else path.lstrip("/").removeprefix("static/")
|
||||
target = (STATIC_DIR / name).resolve()
|
||||
if STATIC_DIR.resolve() not in target.parents or not target.is_file():
|
||||
target = STATIC_DIR / "index.html"
|
||||
body = target.read_bytes()
|
||||
if target.name == "index.html":
|
||||
# Versioned URLs, so no cache (a home-screen app's included) can serve an old client.
|
||||
for asset in (b"/static/app.js", b"/static/styles.css"):
|
||||
body = body.replace(asset + b'"', asset + b"?v=" + VERSION.encode() + b'"')
|
||||
self.send_bytes(body, mimetypes.guess_type(target.name)[0] or "application/octet-stream", "no-cache")
|
||||
|
||||
# --- routes ----------------------------------------------------------
|
||||
|
||||
def do_GET(self) -> None:
|
||||
path = urlparse(self.path).path
|
||||
if path == "/api/content":
|
||||
return self.send_bytes(CONTENT, cache="no-cache")
|
||||
if path == "/api/state":
|
||||
uid = self.identity()
|
||||
if not uid:
|
||||
return self.fail("Not signed in", HTTPStatus.UNAUTHORIZED)
|
||||
with self.store.lock:
|
||||
world, now = self.store.world, self.store.now()
|
||||
player = world["players"].get(uid)
|
||||
if player:
|
||||
player["last_active"] = player["watch"] = now
|
||||
game.process(world, now, self.store.rng)
|
||||
return self.send_json(self.state(uid))
|
||||
if path.startswith("/api/"):
|
||||
return self.fail("Not found", HTTPStatus.NOT_FOUND)
|
||||
self.send_static(path)
|
||||
|
||||
def do_POST(self) -> None:
|
||||
path = urlparse(self.path).path
|
||||
uid = self.identity()
|
||||
if not uid:
|
||||
return self.fail("Not signed in", HTTPStatus.UNAUTHORIZED)
|
||||
body = self.read_body()
|
||||
if body is None:
|
||||
return self.fail("Bad request body")
|
||||
with self.store.lock:
|
||||
world, now = self.store.world, self.store.now()
|
||||
game.process(world, now, self.store.rng)
|
||||
player = world["players"].get(uid)
|
||||
if path == "/api/register":
|
||||
error = self.register(uid, body, now)
|
||||
elif not player:
|
||||
return self.fail("Create your character first", HTTPStatus.CONFLICT)
|
||||
elif path.startswith("/api/dev/"):
|
||||
error = self.dev_route(path, player, body, now) if self.dev else "Not found"
|
||||
else:
|
||||
player["last_active"] = player["watch"] = now
|
||||
error = self.player_route(path, player, body, now)
|
||||
if error:
|
||||
return self.fail(error)
|
||||
game.process(world, now, self.store.rng)
|
||||
if path not in ("/api/press", "/api/target"):
|
||||
self.store.save()
|
||||
self.send_json(self.state(uid))
|
||||
|
||||
def player_route(self, path: str, player: dict, body: dict, now: float) -> str | None:
|
||||
world = self.store.world
|
||||
arg = body.get("id")
|
||||
arg = str(arg) if arg is not None else ""
|
||||
if path == "/api/press":
|
||||
fight = game.current_fight(world, player)
|
||||
if not fight:
|
||||
return "You're not fighting"
|
||||
target = body.get("target")
|
||||
return combat.press(fight, player["id"], arg, now, str(target) if target else None)
|
||||
if path == "/api/target":
|
||||
fight = game.current_fight(world, player)
|
||||
return combat.set_target(fight, player["id"], arg) if fight else "You're not fighting"
|
||||
if path == "/api/manual":
|
||||
return game.set_manual(world, player, arg, bool(body.get("on")))
|
||||
if path == "/api/hunt":
|
||||
return game.start_hunt(world, player, arg, now)
|
||||
if path == "/api/stop":
|
||||
return game.stop(world, player, now)
|
||||
if path == "/api/travel":
|
||||
return game.travel(world, player, arg, now)
|
||||
if path == "/api/equip":
|
||||
return game.equip(player, arg)
|
||||
if path == "/api/unequip":
|
||||
return game.unequip(player, arg)
|
||||
if path == "/api/sell":
|
||||
qty = body.get("qty", 1)
|
||||
return game.sell(player, arg, qty if qty == "all" else (qty if isinstance(qty, int) else -1))
|
||||
if path == "/api/sell_junk":
|
||||
return game.sell_junk(player)
|
||||
if path == "/api/accept":
|
||||
return game.accept(player, arg)
|
||||
if path == "/api/abandon":
|
||||
return game.abandon(player, arg)
|
||||
if path == "/api/turn_in":
|
||||
return game.turn_in(world, player, arg, body.get("choice"), now)
|
||||
if path == "/api/group/create":
|
||||
return game.create_group(world, player)
|
||||
if path == "/api/group/join":
|
||||
return game.join_group(world, player, arg)
|
||||
if path == "/api/group/leave":
|
||||
return game.leave_group(world, player, now)
|
||||
if path == "/api/dungeon/enter":
|
||||
return game.enter_dungeon(world, player, now)
|
||||
if path == "/api/dungeon/pull":
|
||||
return game.pull_now(world, player, now)
|
||||
if path == "/api/dungeon/leave":
|
||||
return game.leave_dungeon(world, player, now)
|
||||
if path == "/api/say":
|
||||
return game.say(world, player, str(body.get("text", "")), now)
|
||||
return "Not found"
|
||||
|
||||
def register(self, uid: str, body: dict, now: float) -> str | None:
|
||||
world = self.store.world
|
||||
if uid in world["players"]:
|
||||
return "You already have a character"
|
||||
name = str(body.get("name", "")).strip().capitalize()
|
||||
cls = body.get("cls")
|
||||
if not NAME.match(name):
|
||||
return "Names are 2 to 16 letters"
|
||||
if any(p["name"].lower() == name.lower() for p in world["players"].values()):
|
||||
return "That name is taken"
|
||||
if cls not in content.CLASSES:
|
||||
return "Pick a class"
|
||||
world["players"][uid] = game.new_player(uid, name, cls, now)
|
||||
game.world_log(world, now, f"{name} the {content.CLASSES[cls]['name']} arrived in Northshire", "join", uid)
|
||||
return None
|
||||
|
||||
def dev_route(self, path: str, player: dict, body: dict, now: float) -> str | None:
|
||||
world = self.store.world
|
||||
if path == "/api/dev/level":
|
||||
return game.dev_level(world, player, int(body.get("level", 1)), now)
|
||||
if path == "/api/dev/advance":
|
||||
world["config"]["clock_offset"] += float(body.get("minutes", 10)) * game.MINUTE
|
||||
return None
|
||||
if path == "/api/dev/teleport":
|
||||
if str(body.get("id")) not in content.PLACES:
|
||||
return "No such place"
|
||||
game.stop(world, player, now)
|
||||
player["place"] = str(body["id"])
|
||||
return None
|
||||
return "Not found"
|
||||
|
||||
def state(self, uid: str) -> dict:
|
||||
world = self.store.world
|
||||
now = self.store.now()
|
||||
player = world["players"].get(uid)
|
||||
fight = game.shown_fight(world, player, now) if player else None
|
||||
group = world["groups"].get(player["group"] or "") if player else None
|
||||
return {
|
||||
"now": now, "dev": self.dev, "uid": uid, "version": VERSION,
|
||||
"me": player_view(world, player, now) if player else None,
|
||||
"fight": combat.view(fight) if fight else None,
|
||||
"group": group_view(world, group, now),
|
||||
"groups": [{"id": g["id"], "leader": world["players"][g["leader"]]["name"], "size": len(g["members"]),
|
||||
"inside": bool(g["dungeon"])} for g in world["groups"].values()],
|
||||
"players": roster(world, now),
|
||||
"log": world["log"][-30:],
|
||||
"chat": world["chat"][-60:],
|
||||
}
|
||||
|
||||
|
||||
def main() -> None:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("--host", default="127.0.0.1")
|
||||
parser.add_argument("--port", type=int, default=8100)
|
||||
parser.add_argument("--db", type=Path, default=Path("wog.sqlite3"))
|
||||
parser.add_argument("--dev", action="store_true", help="trust a wog_user cookie and enable cheats")
|
||||
parser.add_argument("--user-header", default="Remote-User")
|
||||
parser.add_argument("--backup-to", type=Path, help="write a snapshot of the database to this directory and exit")
|
||||
parser.add_argument("--keep", type=int, default=30, help="snapshots to keep with --backup-to")
|
||||
args = parser.parse_args()
|
||||
logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
|
||||
|
||||
store = Store(args.db, args.dev)
|
||||
if args.backup_to:
|
||||
logging.info("Wrote %s", store.backup(args.backup_to, args.keep))
|
||||
return
|
||||
Handler.store = store
|
||||
Handler.dev = args.dev
|
||||
Handler.user_header = args.user_header
|
||||
|
||||
def ticker() -> None:
|
||||
while True:
|
||||
time.sleep(TICK_SECONDS)
|
||||
try:
|
||||
store.tick()
|
||||
except Exception:
|
||||
logging.exception("tick failed")
|
||||
|
||||
threading.Thread(target=ticker, daemon=True).start()
|
||||
server = ThreadingHTTPServer((args.host, args.port), Handler)
|
||||
logging.info("World of Guildhall listening on http://%s:%d%s", args.host, args.port, " (dev)" if args.dev else "")
|
||||
server.serve_forever()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Loading…
Add table
Add a link
Reference in a new issue