#!/usr/bin/env python3 """Spil et helt spil med bots mod en kørende server (røgtest). Kør i CT143 (eller lokalt), hvor værts-headeren stoles på fra 127.0.0.1: /opt/wpp/venv/bin/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. """ import argparse import asyncio import http.cookiejar import json import random import re import sys import urllib.parse import urllib.request import websockets HOST_HEADERS = {"X-authentik-username": "bot-host", "X-authentik-groups": "Familie"} def create_room(base: str) -> str: jar = http.cookiejar.CookieJar() opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(jar)) req = urllib.request.Request(f"{base}/host/", headers=HOST_HEADERS) html = opener.open(req).read().decode() token = re.search(r'name="csrfmiddlewaretoken" value="([^"]+)"', html).group(1) data = urllib.parse.urlencode({"csrfmiddlewaretoken": token}).encode() req = urllib.request.Request(f"{base}/host/new", data=data, headers={**HOST_HEADERS, "Referer": f"{base}/host/"}) resp = opener.open(req) return resp.geturl().rstrip("/").rsplit("/", 1)[-1] 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 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"]) 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): 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']}") continue if msg["type"] != "state": continue st, me = msg["state"], msg["state"]["me"] 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"] == "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"]})) 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) args = ap.parse_args() rng = random.Random(args.seed) code = create_room(args.base) 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"][: args.players] 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))) 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: 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))) try: await asyncio.wait_for(done.wait(), timeout=600) except asyncio.TimeoutError: print("Spillet blev ikke færdigt inden for 10 minutter", file=sys.stderr) return 1 for t in tasks: t.cancel() needed = {"lobby", "category", "lie", "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 if __name__ == "__main__": sys.exit(asyncio.run(main()))