weirsoe-party-protocol/deploy/bots.py

#!/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.
"""

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) -> 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}).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) 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, stop_at_final=False):
    try:
        await _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()


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"] == "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)
    ap.add_argument("--code", help="join et eksisterende rum i stedet for at oprette et (du er selv vært/VIP)")
    args = ap.parse_args()
    rng = random.Random(args.seed)
    if args.code:
        return await join_existing(args, rng)
    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


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"][: args.players]
    tasks = [asyncio.create_task(player(ws_base, code, n, 99, done, rng, stop_at_final=True)) 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()))