#!/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 = game.migrate(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, "zones": content.ZONES, "roads": content.ROADS, "flight_cost": content.FLIGHT_COST, "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", "zones", "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/journey": return game.journey(world, player, arg, now) if path == "/api/fly": return game.fly(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"]) if game.zone_of(player["place"]) not in player["zones"]: player["zones"].append(game.zone_of(player["place"])) 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()