6d1dd5d56c
- server.py: zero-dependency stdlib backend (HTTP + WebSocket /ws), 大话骰 rule engine, server-driven bots - live.html: real MVP frontend wired to the backend over WebSocket - index.html/app.js/style.css: older static prototype - deploy/: systemd unit + Nginx reverse proxy + Let's Encrypt setup.sh - test_*.py: standalone HTTP / in-process test scripts Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
110 lines
4.8 KiB
Python
110 lines
4.8 KiB
Python
import socket, hashlib, base64, struct, json, time, threading, os
|
||
|
||
HOST, PORT = '127.0.0.1', 8765
|
||
|
||
class WS:
|
||
"""极简 WebSocket 客户端（标准库）。客户端帧必须带掩码。"""
|
||
def __init__(self):
|
||
self.sock = socket.create_connection((HOST, PORT))
|
||
key = base64.b64encode(os.urandom(16)).decode()
|
||
req = (f"GET /ws HTTP/1.1\r\nHost: {HOST}:{PORT}\r\nUpgrade: websocket\r\n"
|
||
f"Connection: Upgrade\r\nSec-WebSocket-Key: {key}\r\nSec-WebSocket-Version: 13\r\n\r\n")
|
||
self.sock.sendall(req.encode())
|
||
# 读到握手响应头结束
|
||
buf = b""
|
||
while b"\r\n\r\n" not in buf:
|
||
buf += self.sock.recv(1)
|
||
assert b"101" in buf, "握手失败: "+buf.decode(errors='ignore')
|
||
self.inbox = []
|
||
self._buf = b""
|
||
self.alive = True
|
||
threading.Thread(target=self._reader, daemon=True).start()
|
||
|
||
def _recv_exact(self, n):
|
||
while len(self._buf) < n:
|
||
chunk = self.sock.recv(4096)
|
||
if not chunk: return None
|
||
self._buf += chunk
|
||
out, self._buf = self._buf[:n], self._buf[n:]
|
||
return out
|
||
|
||
def _reader(self):
|
||
while self.alive:
|
||
head = self._recv_exact(2)
|
||
if not head: break
|
||
length = head[1] & 0x7f
|
||
if length == 126: length = struct.unpack(">H", self._recv_exact(2))[0]
|
||
elif length == 127: length = struct.unpack(">Q", self._recv_exact(8))[0]
|
||
payload = self._recv_exact(length) if length else b""
|
||
if payload is None: break
|
||
try: self.inbox.append(json.loads(payload.decode()))
|
||
except Exception: pass
|
||
|
||
def send(self, obj):
|
||
data = json.dumps(obj).encode()
|
||
mask = os.urandom(4)
|
||
masked = bytes(data[i] ^ mask[i % 4] for i in range(len(data)))
|
||
n = len(data); header = bytes([0x81])
|
||
if n < 126: header += bytes([0x80 | n])
|
||
elif n < 65536: header += bytes([0x80 | 126]) + struct.pack(">H", n)
|
||
else: header += bytes([0x80 | 127]) + struct.pack(">Q", n)
|
||
self.sock.sendall(header + mask + masked)
|
||
|
||
def wait(self, pred, timeout=4):
|
||
end = time.time() + timeout
|
||
while time.time() < end:
|
||
for m in list(self.inbox):
|
||
if pred(m): return m
|
||
time.sleep(0.03)
|
||
raise TimeoutError("等待消息超时")
|
||
|
||
def close(self):
|
||
self.alive = False
|
||
try: self.sock.close()
|
||
except Exception: pass
|
||
|
||
# --- 1) 连接 + 初始房间列表 ---
|
||
host = WS()
|
||
host.wait(lambda m: m['type'] == 'rooms')
|
||
print('1. 连接成功，收到初始 rooms')
|
||
|
||
# --- 2) 房主创建房间 ---
|
||
host.send({'type': 'create', 'game': 'dice', 'name': '小鱼', 'playerId': 'p-host'})
|
||
j = host.wait(lambda m: m['type'] == 'joined')
|
||
code = j['room']['code']
|
||
print('2. create -> joined, 房号', code)
|
||
|
||
# --- 3) 第二个客户端加入，房主应收到广播（玩家数变 2） ---
|
||
guest = WS(); guest.wait(lambda m: m['type'] == 'rooms')
|
||
host.inbox.clear()
|
||
guest.send({'type': 'join', 'code': code, 'name': '阿强', 'playerId': 'p-guest'})
|
||
guest.wait(lambda m: m['type'] == 'joined')
|
||
hb = host.wait(lambda m: m['type'] == 'room' and len(m['room']['players']) == 2)
|
||
print('3. 第二人加入，房主收到广播，玩家数 =', len(hb['room']['players']))
|
||
|
||
# 访客离开（广播验证完成），改用「房主 + 3 机器人」验证服务端自驱
|
||
guest.send({'type': 'leave'}); guest.wait(lambda m: m['type'] == 'left')
|
||
|
||
# --- 4) 加 3 机器人并开始；机器人由服务端自动推进 ---
|
||
host.send({'type': 'addBot'}); host.send({'type': 'addBot'}); host.send({'type': 'addBot'})
|
||
host.wait(lambda m: m['type'] == 'room' and len(m['room']['players']) == 4)
|
||
host.inbox.clear()
|
||
host.send({'type': 'start'})
|
||
playing = host.wait(lambda m: m['type'] == 'room' and m['room']['status'] == 'playing')
|
||
print('4. 游戏开始，status =', playing['room']['status'], '首轮 =', playing['room']['turnPlayerName'])
|
||
|
||
# --- 5) 房主叫一手后，其余全是机器人；不发任何 bot 指令，验证服务端自动驱动 ---
|
||
host.inbox.clear()
|
||
host.send({'type': 'call', 'count': 4, 'point': 1})
|
||
|
||
# 等待出现「机器人产生的叫点历史」——证明服务端自驱
|
||
got = host.wait(lambda m: m['type'] == 'room' and any(
|
||
h.get('type') == 'call' and h.get('player') not in ('小鱼', '阿强')
|
||
for h in m['room'].get('history', [])), timeout=6)
|
||
bot_calls = [h['text'] for h in got['room']['history'] if h.get('type') == 'call' and h.get('player') not in ('小鱼', '阿强')]
|
||
print('5. 服务端自动驱动机器人叫点（无需客户端 bot-turn）:', bot_calls[:3])
|
||
assert bot_calls, '机器人未被服务端自动驱动'
|
||
|
||
host.close(); guest.close()
|
||
print('PASS: WebSocket 端到端通过（连接/创建/加入/广播/服务端机器人驱动）')
|