#!/usr/bin/env python3 """Small, dependency-free server for Guildhall, a co-op idle game.""" 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 game APP_DIR = Path(__file__).resolve().parent STATIC_DIR = APP_DIR / "static" MAX_BODY = 16 * 1024 NAME = re.compile(r"^[\w .'-]{2,20}$") TICK_SECONDS = 15 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 / "app.py", APP_DIR / "game.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 after each change.""" def __init__(self, path: Path, speed: float): self.path = path self.lock = threading.Lock() self.rng = random.Random() 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() if row: world = json.loads(row[0]) self.world = game.migrate(world, time.time() + world["config"].get("clock_offset", 0)) self.world["config"]["speed"] = speed else: self.world = game.new_world(time.time(), speed) 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,)) 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"guildhall-{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("guildhall-*.sqlite3")) for old in snapshots[:-keep] if keep > 0 else []: old.unlink() return target def tick(self) -> None: with self.lock: if game.tick(self.world, self.now(), self.rng): self.save() def player_view(world: dict, player: dict, now: float) -> dict: game.regen(world, player, now) view = {key: value for key, value in player.items() if key not in ("last_active", "hp_at")} view["hp"] = int(player["hp"]) view["max_hp"] = game.max_hp(player) view["armour"] = game.armour(player) view["reduction"] = round(game.damage_reduction(player), 3) view["edible"] = game.edible(player) blessed = game.blessing(player) view["blessing"] = blessed view["fight_damage"] = {a["id"]: round(game.fight_damage(player, a, blessed), 1) for a in game.ACTIONS if a["skill"] == "combat"} view["pp"] = int(player["pp"]) view["levels"] = {skill: game.skill_level(player, skill) for skill in game.SKILL_IDS} view["slots"] = game.bank_slots(world, player) view["stack"] = game.stack_size(world) view["power"] = game.combat_power(player) view["perk_costs"] = {perk: game.perk_cost(player, perk) for perk in player["perks"]} if player["action"]: action = game.ACTION_BY_ID[player["action"]["id"]] view["action"] = {**player["action"], "duration": game.action_time(world, player, action)} view["times"] = {action["id"]: game.action_time(world, player, action) for action in game.ACTIONS} view["blocked"] = {action["id"]: game.can_start(world, player, action) for action in game.ACTIONS} return view def city_view(world: dict, now: float) -> dict: city = world["city"] buildings = {} for key, building in city["buildings"].items(): working = city["construction"] and city["construction"]["building"] == key target = game.fund_target(world, key) missing = game.fund_missing(world, key) if working: status = "building" elif key in city["queue"]: status = "queued" elif target is None: status = "max" if building["level"] >= game.BUILDINGS[key]["max"] else "capped" else: status = "funding" buildings[key] = {**building, "target": target, "missing": missing, "status": status, "cap": game.level_cap(world, key)} return { **game.population_view(world), "buildings": buildings, "queue": city["queue"], "devotion": round(city.get("devotion", 0), 1), "ward": game.ward(world), "treasury": int(city["treasury"]), "stockpile": city["stockpile"], "construction": city["construction"], "starving": city["starving"], "merchant_until": city.get("merchant_until", 0), "repelled": city["repelled"], "defence": game.defence(world), "day": int((now - world["created"]) * game.speed(world) // game.DAY) + 1, } def roster(world: dict, now: float) -> list[dict]: """What everyone can see about each other: no banks, no gold.""" people = [] for player in world["players"].values(): action = game.ACTION_BY_ID[player["action"]["id"]] if player["action"] else None game.regen(world, player, now) people.append({ "xp": player["xp"], "equip": player["equip"], "joined": player["joined"], "hp": int(player["hp"]), "max_hp": game.max_hp(player), "power": game.combat_power(player), "armour": game.armour(player), "reduction": round(game.damage_reduction(player), 3), "favour": player["favour"], "id": player["id"], "name": player["name"], "cls": player["cls"], "total": sum(game.skill_level(player, skill) for skill in game.SKILL_IDS), "levels": {skill: game.skill_level(player, skill) for skill in game.SKILL_IDS}, "doing": action["name"] if action else None, "skill": action["skill"] if action else None, "garrison": player["garrison"], "donated": player["donated"], }) return sorted(people, key=lambda p: -p["donated"]) class Handler(BaseHTTPRequestHandler): server_version = "Guildhall" 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*)gh_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_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() self.send_response(HTTPStatus.OK) self.send_header("Content-Type", mimetypes.guess_type(target.name)[0] or "application/octet-stream") self.send_header("Content-Length", str(len(body))) self.send_header("Cache-Control", "no-cache") self.end_headers() self.wfile.write(body) # --- routes ---------------------------------------------------------- def do_GET(self) -> None: path = urlparse(self.path).path if path == "/api/content": self.send_json(game.content()) elif path == "/api/state": uid = self.identity() if not uid: return self.fail("Not signed in", HTTPStatus.UNAUTHORIZED) with self.store.lock: world = self.store.world now = self.store.now() game.tick(world, now, self.store.rng) player = world["players"].get(uid) summary = None if player: summary = game.process_player(world, player, now, self.store.rng) game.regen(world, player, now) player["last_active"] = now self.store.save() self.send_json(self.state(uid, summary)) elif path.startswith("/api/"): self.fail("Not found", HTTPStatus.NOT_FOUND) else: 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 = self.store.world now = self.store.now() game.tick(world, now, self.store.rng) player = world["players"].get(uid) if path == "/api/register": error = self.register(uid, body, now) elif path.startswith("/api/dev/") and self.dev: error = self.dev_route(path, body) elif not player: return self.fail("Create your adventurer first", HTTPStatus.CONFLICT) else: summary = game.process_player(world, player, now, self.store.rng) game.regen(world, player, now) player["last_active"] = now error = self.player_route(path, player, body, now) if error is None: game.tick(world, now, self.store.rng) self.store.save() if error: return self.fail(error) return self.send_json(self.state(uid, summary)) self.store.save() if error: return self.fail(error) self.send_json(self.state(uid)) def player_route(self, path: str, player: dict, body: dict, now: float) -> str | None: world = self.store.world item = body.get("item") qty = body.get("qty") if path in ("/api/donate", "/api/sell", "/api/give"): if item != "gold" and item not in game.ITEMS: return "No such item" if qty == "all": qty = player["gold"] if item == "gold" else player["bank"].get(item, 0) if not isinstance(qty, int): return "Bad quantity" if path == "/api/action": return game.start_action(world, player, str(body.get("id")), now, once=bool(body.get("once"))) if path == "/api/stop": game.stop(player) return None if path == "/api/donate": return game.donate(world, player, item, qty, now) if path == "/api/give": receiver = world["players"].get(str(body.get("to"))) if not receiver: return "No such adventurer" if item == "gold": return "Gold goes to the treasury, not between players" return game.give(world, player, receiver, item, qty, now) if path == "/api/sell": return "Gold can't be sold" if item == "gold" else game.sell(world, player, item, qty, now) if path == "/api/equip": return self.equip(player, item, body.get("slot")) if path == "/api/fund": qty = body.get("qty") return game.fund(world, player, str(body.get("building")), str(body.get("item")), qty if qty == "all" else (qty if isinstance(qty, int) else -1), str(body.get("source")), now) if path == "/api/garrison": raid = game.upcoming_raid(world) if body.get("join") and not raid: return "There's no raid to stand guard for" player["garrison"] = bool(body.get("join")) return None if path == "/api/prayers": ids = body.get("ids") return game.set_prayers(player, ids) if isinstance(ids, list) else "Bad request" if path == "/api/food": food = body.get("item") if food is not None and "heal" not in game.ITEMS.get(str(food), {}): return "That isn't food" player["food"] = food return None if path == "/api/favour": return game.spend_favour(world, player, str(body.get("kind")), now) if path == "/api/perk": return game.buy_perk(player, str(body.get("perk"))) if path == "/api/settings": if body.get("overflow") not in ("city", "stop"): return "Bad setting" player["overflow"] = body["overflow"] return None if path == "/api/board": text = str(body.get("text", "")).strip()[:500] if not text: return "Say something" world["board"].append({"at": now, "who": player["id"], "text": text}) del world["board"][:-game.MAX_LOG] return None return "Not found" def equip(self, player: dict, item: str | None, slot: str | None) -> str | None: if item is None: if slot not in player["equip"]: return "No such slot" current = player["equip"][slot] if current: if game.bank_room(self.store.world, player, current) < 1: return "No room in the bank" player["bank"][current] = player["bank"].get(current, 0) + 1 player["equip"][slot] = None return None spec = game.ITEMS.get(item) if not spec or "slot" not in spec: return "That can't be equipped" if player["bank"].get(item, 0) < 1: return "You don't have one" problem = game.equip_problem(player, item) if problem: return problem game.bank_take(player, item, 1) current = player["equip"][spec["slot"]] if current: if game.bank_room(self.store.world, player, current) < 1: player["bank"][item] = player["bank"].get(item, 0) + 1 return "No room in the bank for what you're wearing" player["bank"][current] = player["bank"].get(current, 0) + 1 player["equip"][spec["slot"]] = item return None def register(self, uid: str, body: dict, now: float) -> str | None: world = self.store.world if uid in world["players"]: return "You already have an adventurer" name = str(body.get("name", "")).strip() cls = body.get("cls") if not NAME.match(name): return "Pick a name of 2 to 20 letters" if cls not in game.CLASSES: return "Pick a class" world["players"][uid] = game.new_player(uid, name, cls, now) game.log(world, now, "join", f"{name} the {game.CLASSES[cls]['name'].lower()} arrived in the city", uid) return None def dev_route(self, path: str, body: dict) -> str | None: world = self.store.world if path == "/api/dev/advance": hours = float(body.get("hours", 1)) world["config"]["clock_offset"] += hours * game.HOUR / game.speed(world) game.tick(world, self.store.now(), self.store.rng) return None if path == "/api/dev/event": world["next_event_at"] = self.store.now() game.tick(world, self.store.now(), self.store.rng) return None return "Not found" def state(self, uid: str, summary: dict | None = None) -> dict: world = self.store.world now = self.store.now() player = world["players"].get(uid) upcoming = game.upcoming_event(world) history = [ {"id": e["id"], "kind": e["kind"], "at": e["resolved"], "report": e.get("report")} for e in reversed(world["events"]) if e["status"] == "done" ][:10] return { "now": now, "dev": self.dev, "uid": uid, "version": VERSION, "me": player_view(world, player, now) if player else None, "summary": summary, "city": city_view(world, now), "upcoming": game.event_view(world, upcoming, now) if upcoming else None, "favours": game.favours(world, now), "history": history, "players": roster(world, now), "ledger": world["ledger"][-80:], "board": world["board"][-80:], } def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--host", default="127.0.0.1") parser.add_argument("--port", type=int, default=8098) parser.add_argument("--db", type=Path, default=Path("guildhall.sqlite3")) parser.add_argument("--speed", type=float, default=1.0, help="game time multiplier (dev worlds)") parser.add_argument("--dev", action="store_true", help="trust a gh_user cookie and enable time travel") 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.speed) 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("Guildhall listening on http://%s:%d%s", args.host, args.port, " (dev)" if args.dev else "") server.serve_forever() if __name__ == "__main__": main()