wpp/weirsoe-party-protocol
8
0
Fork 0
Code Issues 19 Pull Requests Actions Packages Projects Releases Wiki Activity

Bots: --game tegn spiller et helt Tegn og Gæt

Browse Source 
Bots tegner tilfældige streger, gætter tilfældige ord, bruger af og
til forslaget og giver hjerter under afsløringen.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
agw
2026-10-20 17:33:07 +02:00
co-authored by Claude Opus 5.5
parent 6c60d56409
commit a3310c2325
1 changed files with 91 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
deploy/bots.py
+91 -11
View File
@@ -7,6 +7,10 @@ Kør på samme maskine som serveren, hvor værts-headeren stoles på fra 127.0.0
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.
"""
import argparse
@@ -24,13 +28,13 @@ import websockets
HOST_HEADERS = {"X-authentik-username": "bot-host", "X-authentik-groups": "Familie"}
def create_room(base: str, pack: str = "alle") -> str:
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}).encode()
data = urllib.parse.urlencode({"csrfmiddlewaretoken": token, "pack": pack, "game": game}).encode()
class NoRedirect(urllib.request.HTTPRedirectHandler):
def redirect_request(self, *a, **k):
@@ -48,7 +52,7 @@ def create_room(base: str, pack: str = "alle") -> str:
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) as ws:
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":
@@ -68,9 +72,9 @@ async def host(ws_base, code, seen: list, done: asyncio.Event):
return
async def player(ws_base, code, name, n_players, done: asyncio.Event, rng: random.Random, stop_at_final=False):
async def player(ws_base, code, name, n_players, done: asyncio.Event, rng: random.Random, stop_at_final=False, game="fof"):
try:
await _player(ws_base, code, name, n_players, done, rng, stop_at_final)
await (_tegn_player if game == "tegn" else _player)(ws_base, code, name, n_players, done, rng, stop_at_final)
except websockets.ConnectionClosed:
print(f"[{name}] forbindelsen blev lukket (findes rummet?)")
done.set()
@@ -128,6 +132,80 @@ async def _player(ws_base, code, name, n_players, done, rng, stop_at_final):
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"))
await asyncio.sleep(rng.uniform(0.05, 0.4))
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"] == "theme" and me.get("can_choose") and key not in acted:
acted.add(key)
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)
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 ["😂"])}))
async def main():
ap = argparse.ArgumentParser()
ap.add_argument("--base", default="http://127.0.0.1:8000")
@@ -136,11 +214,12 @@ async def main():
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"], help="spillet (fof = Fup og Fakta, tegn = Tegn og Gæt)")
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)
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()
@@ -148,15 +227,15 @@ async def main():
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)))
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:
if "category" in seen or "theme" 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)))
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=600)
except asyncio.TimeoutError:
@@ -164,7 +243,8 @@ async def main():
return 1
for t in tasks:
t.cancel()
needed = {"lobby", "rules", "category", "lie", "present", "guess", "reveal", "scores", "final"}
needed = ({"lobby", "rules", "theme", "word", "draw", "guess", "reveal", "final"} if args.game == "tegn" else
{"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)
@@ -178,7 +258,7 @@ async def join_existing(args, rng):
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"][: args.players]
tasks = [asyncio.create_task(player(ws_base, code, n, 99, done, rng, stop_at_final=True)) for n in names]
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)