agwandClaude Opus 5.5
5b24404eac
Replikker:
- To spioner siges i flertal ("Spionerne var ..."), også i den generiske
udgave; fanges den ene, og den anden slipper, er det en halv sejr
("Den ene spion er fanget, men den anden slap væk"), ikke "Agenterne
vinder". Storskærmens overskrift følger med.
- Er spionen gået i løbet af runden, siges "Spionen forlod missionen", og
navnet står stadig i afsløringen (motoren husker spionernes navne fra
rundens start; storskærmen viser dem uden farve).
- Fire små replikker skrevet om (vedkommende, "Måske er det ansigtet",
skarpeste øje = flest rigtige stemmer, vinderen gentog ikke den faste).
Ord: Spionen bruger ikke ord under tre bogstaver ("is", "fe"); de står inde
i for mange andre ord (isvaffel), til at stikordene kan holdes fri af dem.
Mad har 17 ord, Eventyr 15.
Test: speaker-testen kører uden rigtige ventetider (37 s til 4 s for hele
filen); skift-testen venter på det nye spils lobby i stedet for at gætte
på tiden.
deploy/bots.py --game spion: tilfældige stikord (prøver et andet, hvis det
afvises), tilfældige stemmer og gæt; værten forkorter snakken. En stemme,
der kommer et øjeblik efter, at fasen sluttede, er ikke en fejl.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
347 lines
18 KiB
Python
347 lines
18 KiB
Python
#!/usr/bin/env python3
|
|
"""Spil et helt spil med bots mod en kørende server (røgtest).
|
|
|
|
Kør på samme maskine som serveren, hvor værts-headeren stoles på fra 127.0.0.1:
|
|
python deploy/bots.py --base http://127.0.0.1:8000 --players 3 --audience 1
|
|
|
|
Opretter et rum som vært, joiner N spillere og M publikummer, og lader dem lyve,
|
|
gætte og like tilfældigt, indtil spillet er slut. Afslutter med kode 0, hvis alle
|
|
faser blev nået og slutstillingen blev vist.
|
|
|
|
Tegn og Gæt (mindst 3 spillere):
|
|
python deploy/bots.py --base http://127.0.0.1:8000 --game tegn --players 5
|
|
Bots tegner tilfældige streger og gætter tilfældige ord.
|
|
|
|
Spionen (mindst 4 spillere):
|
|
python deploy/bots.py --base http://127.0.0.1:8000 --game spion --players 10
|
|
Bots giver tilfældige stikord, stemmer tilfældigt, og en anklaget spion gætter tilfældigt. Værten forkorter snakken.
|
|
"""
|
|
|
|
import argparse
|
|
import asyncio
|
|
import json
|
|
import random
|
|
import re
|
|
import sys
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
|
|
import websockets
|
|
|
|
HOST_HEADERS = {"X-authentik-username": "bot-host", "X-authentik-groups": "Familie"}
|
|
|
|
|
|
def create_room(base: str, pack: str = "alle", game: str = "fof") -> str:
|
|
# CSRF-cookien er Secure i drift; over http til localhost sender vi den selv tilbage.
|
|
resp = urllib.request.urlopen(urllib.request.Request(f"{base}/host/", headers=HOST_HEADERS))
|
|
html = resp.read().decode()
|
|
cookie = re.search(r"csrftoken=([^;]+)", resp.headers.get("Set-Cookie", "")).group(1)
|
|
token = re.search(r'name="csrfmiddlewaretoken" value="([^"]+)"', html).group(1)
|
|
data = urllib.parse.urlencode({"csrfmiddlewaretoken": token, "pack": pack, "game": game}).encode()
|
|
|
|
class NoRedirect(urllib.request.HTTPRedirectHandler):
|
|
def redirect_request(self, *a, **k):
|
|
return None
|
|
|
|
req = urllib.request.Request(f"{base}/host/new", data=data, headers={
|
|
**HOST_HEADERS, "Referer": f"{base}/host/", "Cookie": f"csrftoken={cookie}"})
|
|
try:
|
|
urllib.request.build_opener(NoRedirect).open(req)
|
|
except urllib.error.HTTPError as e:
|
|
if e.code != 302:
|
|
raise
|
|
return e.headers["Location"].rstrip("/").rsplit("/", 1)[-1]
|
|
raise RuntimeError("forventede redirect til værtssiden")
|
|
|
|
|
|
async def host(ws_base, code, seen: list, done: asyncio.Event):
|
|
async with websockets.connect(f"{ws_base}/ws/host/{code}/", additional_headers=HOST_HEADERS, max_size=None) as ws:
|
|
async for raw in ws:
|
|
msg = json.loads(raw)
|
|
if msg["type"] != "state":
|
|
continue
|
|
st = msg["state"]
|
|
if not seen or seen[-1] != st["phase"]:
|
|
seen.append(st["phase"])
|
|
if st.get("game") == "spion" and st["phase"] == "talk": # værten forkorter snakken (bots snakker ikke)
|
|
await ws.send(json.dumps({"action": "skip"}))
|
|
extra = st.get("question", {}).get("prompt", "") if st["phase"] == "lie" else ""
|
|
print(f"[vært] {st['phase']:8} {extra}")
|
|
if st["phase"] == "final":
|
|
for r in st["final"]["ranking"]:
|
|
name = next(p["name"] for p in st["players"] if p["id"] == r["player_id"])
|
|
print(f" {r['place']}. {name} {r['score']}")
|
|
for a in st["final"]["awards"]:
|
|
print(f" {a['title']}: {a['detail']}")
|
|
done.set()
|
|
return
|
|
|
|
|
|
async def player(ws_base, code, name, n_players, done: asyncio.Event, rng: random.Random, stop_at_final=False, game="fof"):
|
|
try:
|
|
play = {"tegn": _tegn_player, "spion": _spion_player}.get(game, _player)
|
|
await play(ws_base, code, name, n_players, done, rng, stop_at_final)
|
|
except websockets.ConnectionClosed:
|
|
print(f"[{name}] forbindelsen blev lukket (findes rummet?)")
|
|
done.set()
|
|
|
|
|
|
async def _player(ws_base, code, name, n_players, done, rng, stop_at_final):
|
|
async with websockets.connect(f"{ws_base}/ws/play/{code}/") as ws:
|
|
await ws.send(json.dumps({"action": "join", "name": name}))
|
|
acted = set()
|
|
while not done.is_set():
|
|
try:
|
|
raw = await asyncio.wait_for(ws.recv(), timeout=1)
|
|
except asyncio.TimeoutError:
|
|
continue
|
|
msg = json.loads(raw)
|
|
if msg["type"] == "error":
|
|
print(f"[{name}] fejl: {msg['code']} {msg['message']}")
|
|
if msg["code"] == "unknown_room":
|
|
done.set()
|
|
return
|
|
continue
|
|
if msg["type"] != "state":
|
|
continue
|
|
st, me = msg["state"], msg["state"]["me"]
|
|
if st["phase"] == "final" and stop_at_final:
|
|
print(f"[{name}] slut: plads {next((r['place'] for r in st['final']['ranking'] if r['player_id'] == me['id']), '-')}")
|
|
done.set()
|
|
return
|
|
key = (st["phase"], (st.get("round") or {}).get("number"), (st.get("round") or {}).get("question_number"))
|
|
await asyncio.sleep(rng.uniform(0.05, 0.4))
|
|
if st["phase"] in ("present", "guess", "reveal") and rng.random() < 0.25:
|
|
for _ in range(rng.randint(1, 8)): # emojis må spammes; serveren drosler
|
|
await ws.send(json.dumps({"action": "react", "emoji": rng.choice(st.get("reactions") or ["😂", "😱", "👏", "🤯", "🙈", "🔥"])}))
|
|
if st["phase"] == "lobby" and me["vip"] and len(st["players"]) >= n_players and key not in acted:
|
|
acted.add(key)
|
|
await ws.send(json.dumps({"action": "start"}))
|
|
elif st["phase"] == "category" and me["is_chooser"] and key not in acted:
|
|
acted.add(key)
|
|
await ws.send(json.dumps({"action": "choose_category", "category_id": st["category"]["options"][0]["id"]}))
|
|
elif st["phase"] == "lie" and me["role"] == "player" and not me["lie"] and key not in acted:
|
|
acted.add(key)
|
|
if rng.random() < 0.3:
|
|
await ws.send(json.dumps({"action": "lie_for_me"}))
|
|
sug = json.loads(await ws.recv())
|
|
text = sug.get("text") if sug.get("type") == "suggestion" else None
|
|
else:
|
|
text = None
|
|
await ws.send(json.dumps({"action": "submit_lie", "text": text or f"{name}s løgn {rng.randint(1, 999)}"}))
|
|
elif st["phase"] == "guess" and not me["guess"] and key not in acted:
|
|
acted.add(key)
|
|
answer = rng.choice(st["answers"])
|
|
await ws.send(json.dumps({"action": "guess", "answer_id": answer["id"]}))
|
|
others = [a for a in st["answers"] if a["id"] != answer["id"] and a["id"] != me.get("own_answer_id")]
|
|
if others and rng.random() < 0.5:
|
|
await ws.send(json.dumps({"action": "like", "answer_id": rng.choice(others)["id"]}))
|
|
|
|
|
|
TEGN_WORDS = ["kat", "hund", "hus", "sol", "bil", "træ", "fisk", "pizza", "raket", "blomst", "sofa", "and"]
|
|
PALETTE = ["#1d1b2e", "#e63946", "#2a7de1", "#2bb673"]
|
|
|
|
|
|
def random_drawing(rng: random.Random) -> list[dict]:
|
|
"""Nogle tilfældige streger (random walk) i lærredets normaliserede koordinater."""
|
|
strokes = []
|
|
for _ in range(rng.randint(1, 6)):
|
|
x, y, pts = rng.random(), rng.random(), []
|
|
for _ in range(rng.randint(1, 30)):
|
|
x = min(1, max(0, x + rng.uniform(-0.05, 0.05)))
|
|
y = min(1, max(0, y + rng.uniform(-0.05, 0.05)))
|
|
pts.append([round(x, 3), round(y, 3)])
|
|
strokes.append({"color": rng.choice(PALETTE), "width": rng.choice([0.008, 0.024]), "points": pts})
|
|
return strokes
|
|
|
|
|
|
async def _tegn_player(ws_base, code, name, n_players, done, rng, stop_at_final):
|
|
async with websockets.connect(f"{ws_base}/ws/play/{code}/", max_size=None) as ws:
|
|
await ws.send(json.dumps({"action": "join", "name": name}))
|
|
acted = set()
|
|
while not done.is_set():
|
|
try:
|
|
raw = await asyncio.wait_for(ws.recv(), timeout=1)
|
|
except asyncio.TimeoutError:
|
|
continue
|
|
msg = json.loads(raw)
|
|
if msg["type"] == "error":
|
|
print(f"[{name}] fejl: {msg['code']} {msg['message']}")
|
|
if msg["code"] == "unknown_room":
|
|
done.set()
|
|
return
|
|
continue
|
|
if msg["type"] == "suggestion":
|
|
await ws.send(json.dumps({"action": "submit_word", "text": msg["text"]}))
|
|
continue
|
|
if msg["type"] != "state":
|
|
continue
|
|
st, me = msg["state"], msg["state"]["me"]
|
|
if st["phase"] == "final" and stop_at_final:
|
|
done.set()
|
|
return
|
|
task = me.get("task")
|
|
key = (st["phase"], (st.get("step") or {}).get("index"))
|
|
# Tænkepausen kun før en handling: under afsløringen kommer der mange snapshots, og en bot,
|
|
# der sover efter hvert, når ikke at læse dem (og serverens ping) i tide.
|
|
pause = lambda: asyncio.sleep(rng.uniform(0.05, 0.4)) # noqa: E731
|
|
if st["phase"] == "lobby" and me["vip"] and len(st["players"]) >= n_players and key not in acted:
|
|
acted.add(key)
|
|
await pause()
|
|
await ws.send(json.dumps({"action": "start"}))
|
|
elif st["phase"] == "theme" and me.get("can_choose") and key not in acted:
|
|
acted.add(key)
|
|
await pause()
|
|
await ws.send(json.dumps({"action": "choose_theme", "theme_id": rng.choice(st["theme"]["options"])["id"]}))
|
|
elif task and not task["done"] and key not in acted:
|
|
acted.add(key)
|
|
await pause()
|
|
if task["kind"] == "word":
|
|
if rng.random() < 0.3:
|
|
await ws.send(json.dumps({"action": "suggest"})) # svaret afleveres, når det kommer
|
|
else:
|
|
await ws.send(json.dumps({"action": "submit_word", "text": rng.choice(TEGN_WORDS)}))
|
|
elif task["kind"] == "draw":
|
|
drawing = random_drawing(rng)
|
|
await ws.send(json.dumps({"action": "draft", "strokes": drawing[:1]}))
|
|
await ws.send(json.dumps({"action": "submit_drawing", "strokes": drawing}))
|
|
else:
|
|
await ws.send(json.dumps({"action": "submit_guess", "text": rng.choice(TEGN_WORDS)}))
|
|
elif st["phase"] == "reveal" and st.get("reveal") and rng.random() < 0.3:
|
|
mine = set(me.get("likes") or [])
|
|
options = [e["id"] for c in st["reveal"]["chains"] for e in c["entries"]
|
|
if e["kind"] in ("drawing", "guess") and e["by"] != me["id"] and e["id"] not in mine]
|
|
if options:
|
|
await ws.send(json.dumps({"action": "like", "entry_id": rng.choice(options)}))
|
|
if st["phase"] == "reveal" and rng.random() < 0.2:
|
|
await ws.send(json.dumps({"action": "react", "emoji": rng.choice(st.get("reactions") or ["😂"])}))
|
|
|
|
|
|
SPION_CLUES = ["rund", "lille", "stor", "blød", "hård", "rød", "grøn", "kold", "varm", "hurtig", "langsom", "farlig",
|
|
"sjov", "gammel", "ny", "tung", "let", "lang", "kort", "våd", "tør", "glat", "skarp", "larm", "stille",
|
|
"dyr", "billig", "udenfor", "indenfor", "sommer", "vinter", "morgen", "aften", "hjem", "skole"]
|
|
|
|
|
|
async def _spion_player(ws_base, code, name, n_players, done, rng, stop_at_final):
|
|
async with websockets.connect(f"{ws_base}/ws/play/{code}/") as ws:
|
|
await ws.send(json.dumps({"action": "join", "name": name}))
|
|
acted, state = set(), None
|
|
while not done.is_set():
|
|
try:
|
|
raw = await asyncio.wait_for(ws.recv(), timeout=1)
|
|
except asyncio.TimeoutError:
|
|
continue
|
|
msg = json.loads(raw)
|
|
if msg["type"] == "error":
|
|
if msg["code"] in ("clue_is_word", "clue_is_theme", "clue_taken", "clue_invalid") and state:
|
|
# afvist stikord: prøv et andet (det er en del af spillet, ikke en fejl)
|
|
await ws.send(json.dumps({"action": "submit_clue", "text": rng.choice(SPION_CLUES)}))
|
|
continue
|
|
if msg["code"] == "wrong_phase":
|
|
continue # stemte et øjeblik efter, at alle andre havde stemt: helt normalt
|
|
print(f"[{name}] fejl: {msg['code']} {msg['message']}")
|
|
if msg["code"] == "unknown_room":
|
|
done.set()
|
|
return
|
|
continue
|
|
if msg["type"] != "state":
|
|
continue
|
|
state = st = msg["state"]
|
|
me = st["me"]
|
|
if st["phase"] == "final" and stop_at_final:
|
|
done.set()
|
|
return
|
|
key = (st["phase"], (st.get("round") or {}).get("number"))
|
|
pause = lambda: asyncio.sleep(rng.uniform(0.05, 0.4)) # noqa: E731
|
|
if st["phase"] == "lobby" and me["vip"] and len(st["players"]) >= n_players and key not in acted:
|
|
acted.add(key)
|
|
await pause()
|
|
await ws.send(json.dumps({"action": "start"}))
|
|
elif st["phase"] == "clue" and me.get("card") and not me.get("clue") and key not in acted:
|
|
acted.add(key)
|
|
await pause()
|
|
await ws.send(json.dumps({"action": "submit_clue", "text": rng.choice(SPION_CLUES)}))
|
|
elif st["phase"] == "vote" and not me.get("vote") and key not in acted:
|
|
options = [p["id"] for p in st["players"] if p["in_round"] and p["id"] != me["id"]]
|
|
if options:
|
|
acted.add(key)
|
|
await pause()
|
|
await ws.send(json.dumps({"action": "vote", "player_id": rng.choice(options)}))
|
|
elif st["phase"] == "guess" and me.get("can_guess") and key not in acted:
|
|
acted.add(key)
|
|
await pause()
|
|
await ws.send(json.dumps({"action": "spy_guess", "text": rng.choice(SPION_CLUES)}))
|
|
if st["phase"] in ("talk", "result") and rng.random() < 0.2:
|
|
await ws.send(json.dumps({"action": "react", "emoji": rng.choice(st.get("reactions") or ["😂"])}))
|
|
|
|
|
|
async def main():
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("--base", default="http://127.0.0.1:8000")
|
|
ap.add_argument("--players", type=int, default=3)
|
|
ap.add_argument("--audience", type=int, default=1)
|
|
ap.add_argument("--seed", type=int, default=None)
|
|
ap.add_argument("--pack", default="alle", help="spørgsmålspakke (alle, familie, nørd, arbejde, fest)")
|
|
ap.add_argument("--code", help="join et eksisterende rum i stedet for at oprette et (du er selv vært/VIP)")
|
|
ap.add_argument("--game", default="fof", choices=["fof", "tegn", "spion"],
|
|
help="spillet (fof = Fup og Fakta, tegn = Tegn og Gæt, spion = Spionen)")
|
|
args = ap.parse_args()
|
|
rng = random.Random(args.seed)
|
|
if args.code:
|
|
return await join_existing(args, rng)
|
|
code = create_room(args.base, args.pack, args.game)
|
|
print(f"Rum {code}")
|
|
ws_base = args.base.replace("http", "ws", 1)
|
|
seen, done = [], asyncio.Event()
|
|
names = ["Asger", "Sus", "Jørn", "Mette", "Kaj", "Bodil", "Ib", "Lis", "Vagn", "Inge"][: args.players] # Tegn og Gæt: op til 10
|
|
tasks = [asyncio.create_task(host(ws_base, code, seen, done))]
|
|
await asyncio.sleep(0.2)
|
|
for n in names:
|
|
tasks.append(asyncio.create_task(player(ws_base, code, n, args.players, done, rng, game=args.game)))
|
|
await asyncio.sleep(0.1)
|
|
# Publikum joiner, når spillet er i gang (i lobbyen ville de blive spillere).
|
|
for _ in range(200):
|
|
if "category" in seen or "theme" in seen or "clue" in seen:
|
|
break
|
|
await asyncio.sleep(0.1)
|
|
for i in range(args.audience):
|
|
tasks.append(asyncio.create_task(player(ws_base, code, f"Publikum{i + 1}", args.players, done, rng, game=args.game)))
|
|
try:
|
|
await asyncio.wait_for(done.wait(), timeout=1200) # Tegn og Gæt med 10 spillere tager knap 9 minutter
|
|
except asyncio.TimeoutError:
|
|
print("Spillet blev ikke færdigt inden for 20 minutter", file=sys.stderr)
|
|
return 1
|
|
for t in tasks:
|
|
t.cancel()
|
|
needed = {"tegn": {"lobby", "rules", "theme", "word", "draw", "guess", "reveal", "final"},
|
|
"spion": {"lobby", "rules", "clue", "clues", "talk", "vote", "verdict", "result", "final"},
|
|
}.get(args.game, {"lobby", "rules", "category", "lie", "present", "guess", "reveal", "scores", "final"})
|
|
missing = needed - set(seen)
|
|
if missing:
|
|
print(f"Faser der aldrig blev nået: {missing}", file=sys.stderr)
|
|
return 1
|
|
print("OK: helt spil gennemført")
|
|
return 0
|
|
|
|
|
|
async def join_existing(args, rng):
|
|
"""Bots fylder op i dit eget spil. De starter ikke spillet (det gør VIP'en, altså dig)."""
|
|
ws_base = args.base.replace("http", "ws", 1)
|
|
code, done = args.code.upper(), asyncio.Event()
|
|
names = ["Bot-Bodil", "Bot-Kaj", "Bot-Lis", "Bot-Ib", "Bot-Mette", "Bot-Jørn", "Bot-Sus", "Bot-Vagn",
|
|
"Bot-Inge"][: args.players]
|
|
tasks = [asyncio.create_task(player(ws_base, code, n, 99, done, rng, stop_at_final=True, game=args.game)) for n in names]
|
|
print(f"{len(names)} bots joiner {code}. Start spillet fra din telefon eller storskærmen.")
|
|
try:
|
|
await asyncio.wait_for(done.wait(), timeout=3600)
|
|
except asyncio.TimeoutError:
|
|
return 1
|
|
for t in tasks:
|
|
t.cancel()
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(asyncio.run(main()))
|