82 lines
2.5 KiB
Python
82 lines
2.5 KiB
Python
"""随从真人玩家：加入指定房间占 2 号位并自动行动到对局结束。用法: uv run python side_player.py <房间号>"""
|
|
import asyncio
|
|
import json
|
|
import random
|
|
import sys
|
|
|
|
import websockets
|
|
|
|
sys.stdout.reconfigure(encoding="utf-8")
|
|
random.seed(int(sys.argv[2]) if len(sys.argv) > 2 else 42)
|
|
CODE = sys.argv[1]
|
|
|
|
|
|
class Client:
|
|
def __init__(self, name):
|
|
self.name = name
|
|
self.snap = None
|
|
self.token = None
|
|
|
|
async def connect(self, code):
|
|
self.ws = await websockets.connect(f"ws://127.0.0.1:8000/ws/{code}?name={self.name}")
|
|
asyncio.create_task(self._recv())
|
|
|
|
async def _recv(self):
|
|
try:
|
|
async for raw in self.ws:
|
|
data = json.loads(raw)
|
|
if data["type"] == "snapshot":
|
|
self.snap = data["data"]
|
|
if self.snap.get("token"):
|
|
self.token = self.snap["token"]
|
|
except Exception:
|
|
pass
|
|
|
|
async def send(self, obj):
|
|
await self.ws.send(json.dumps(obj, ensure_ascii=False))
|
|
|
|
def pending(self):
|
|
return (self.snap or {}).get("pending")
|
|
|
|
def phase(self):
|
|
return (self.snap or {}).get("phase")
|
|
|
|
|
|
async def act(c):
|
|
p = c.pending()
|
|
if not p:
|
|
return False
|
|
key = p["key"]
|
|
if p.get("input_text"):
|
|
await c.send({"type": "speech", "text": "我先听大家的分析，保持好人视角推理。"})
|
|
return True
|
|
if key == "elect":
|
|
await c.send({"type": "action", "action": "elect", "target": random.choice([1, None])})
|
|
return True
|
|
options = p.get("options") or []
|
|
if p.get("allow_skip") and random.random() < 0.2:
|
|
await c.send({"type": "action", "action": key, "target": None})
|
|
return True
|
|
pool = options or [None]
|
|
await c.send({"type": "action", "action": key, "target": random.choice(pool)})
|
|
return True
|
|
|
|
|
|
async def main():
|
|
c = Client("随从")
|
|
await c.connect(CODE)
|
|
await asyncio.sleep(0.8)
|
|
me = (c.snap or {}).get("me") or {}
|
|
print(f"已加入 {CODE}，座位 {me.get('seat')}", flush=True)
|
|
deadline = asyncio.get_event_loop().time() + 600
|
|
while asyncio.get_event_loop().time() < deadline:
|
|
await act(c)
|
|
if c.phase() == "game_over":
|
|
break
|
|
await asyncio.sleep(0.5)
|
|
print("对局结束:", c.snap.get("winner_text"), "| 我的身份:",
|
|
next((p["role_name"] for p in c.snap["players"] if p.get("is_me")), "?"), flush=True)
|
|
|
|
|
|
asyncio.run(main())
|