527 lines
23 KiB
Python
527 lines
23 KiB
Python
#!/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)
|
|
view["blessing"] = game.blessing(player)
|
|
view["fight_odds"] = {a["id"]: game.fight_odds(player, a) for a in game.ACTIONS if "enemy" in a}
|
|
view["pp"] = int(player["pp"])
|
|
view["labour_per_action"] = game.labour_per_action(player)
|
|
view["levels"] = {skill: game.skill_level(player, skill) for skill in game.SKILL_IDS}
|
|
view["slots"] = game.bank_slots(world, player)
|
|
view["slots_used"] = game.slots_used(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, fight = game.growing(player), player["action"].get("fight")
|
|
duration = fight["duration"] if fight else game.action_time(world, player, action)
|
|
view["action"] = {**player["action"], "growing": action["id"], "duration": duration}
|
|
# A fight takes as long as it takes: show the average time to a kill.
|
|
view["times"] = {action["id"]: (view["fight_odds"][action["id"]]["kill_time"] + game.RESPAWN) / game.speed(world)
|
|
if "enemy" in action else 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}
|
|
view["drill_per_action"] = game.drill_per_action(player)
|
|
view["recon_gain"] = {action["id"]: game.recon_gain(world, player, action) for action in game.ACTIONS
|
|
if action.get("special") == "recon"}
|
|
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),
|
|
"townsfolk_labour": round(city["population"] * game.TOWNSFOLK_LABOUR, 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": game.calendar_day(world, now) + 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()
|
|
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_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/install":
|
|
return game.install(world, player, str(body.get("fitting")), 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/board/delete":
|
|
message = next((m for m in world["board"] if m.get("id") == body.get("id")), None)
|
|
if not message:
|
|
return "That message is gone"
|
|
if message["who"] != player["id"]:
|
|
return "You can only delete your own messages"
|
|
world["board"].remove(message)
|
|
return None
|
|
if path == "/api/aid":
|
|
return game.aid_problem(world, player, str(body.get("kind")), str(body.get("item")), str(body.get("source")), now)
|
|
if path == "/api/hire":
|
|
return game.hire_mercenaries(world, player, now)
|
|
if path == "/api/potion":
|
|
return game.use_potion(player, str(body.get("item")))
|
|
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/class":
|
|
cls = body.get("cls")
|
|
if not player.get("repick"):
|
|
return "You've already chosen your trade"
|
|
if cls not in game.CLASSES:
|
|
return "Pick a class"
|
|
player["cls"], player["repick"] = cls, False
|
|
game.log(world, now, "join", f"{player['name']} took up the {game.CLASSES[cls]['name'].lower()}'s trade", player["id"])
|
|
return None
|
|
if path == "/api/plot":
|
|
return game.plant_second_plot(world, player, body.get("id") or 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["next_message_id"] = world.get("next_message_id", 1) + 1
|
|
world["board"].append({"id": world["next_message_id"] - 1, "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
|
|
shield = player["equip"].get("shield")
|
|
if spec.get("two_handed") and shield:
|
|
# Both hands on the weapon: the shield goes back to the bank.
|
|
if game.bank_room(self.store.world, player, shield) < 1:
|
|
return "No room in the bank for your shield"
|
|
player["bank"][shield] = player["bank"].get(shield, 0) + 1
|
|
player["equip"]["shield"] = None
|
|
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/happen":
|
|
track = body.get("track")
|
|
fire = {"minor": game._minor_threat, "hardship": game._hardship, "fortune": game._fortune}.get(track)
|
|
if not fire:
|
|
return "No such track"
|
|
fire(world, self.store.now(), self.store.rng, body.get("kind"))
|
|
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"], "track": e.get("track", "major"), "at": e["resolved"], "report": e.get("report")}
|
|
for e in reversed(world["events"]) if e["status"] == "done"
|
|
][:15]
|
|
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),
|
|
"happenings": game.happenings_view(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()
|