Guildhall alpha: co-op idle game
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
commit
9058039eb7
14 changed files with 2815 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 Guildhall, a co-op idle game."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
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
|
||||
|
||||
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:
|
||||
self.world = json.loads(row[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:
|
||||
view = {key: value for key, value in player.items() if key != "last_active"}
|
||||
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():
|
||||
spec = game.BUILDINGS[key]
|
||||
if building["damaged"]:
|
||||
cost = game.repair_cost(key, building["level"])
|
||||
elif building["level"] < spec["max"]:
|
||||
cost = game.build_cost(key, building["level"] + 1)
|
||||
else:
|
||||
cost = None
|
||||
buildings[key] = {**building, "cost": cost, "missing": game.missing_for(city, cost) if cost else []}
|
||||
return {
|
||||
"population": int(city["population"]),
|
||||
"housing": game.housing(world),
|
||||
"treasury": int(city["treasury"]),
|
||||
"stockpile": city["stockpile"],
|
||||
"buildings": buildings,
|
||||
"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) -> list[dict]:
|
||||
people = []
|
||||
for player in world["players"].values():
|
||||
action = game.ACTION_BY_ID[player["action"]["id"]] if player["action"] else None
|
||||
people.append({
|
||||
"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)
|
||||
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)
|
||||
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)
|
||||
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/build":
|
||||
return game.commission(world, player, str(body.get("building")), 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/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"
|
||||
game.bank_take(player, item, 1)
|
||||
current = player["equip"][spec["slot"]]
|
||||
if current:
|
||||
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,
|
||||
"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,
|
||||
"history": history,
|
||||
"players": roster(world),
|
||||
"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()
|
||||
Loading…
Add table
Add a link
Reference in a new issue