Initial commit: 酒桌派对 / 大话骰 multiplayer drinking-game hub

- 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>
This commit is contained in:
gongch
2026-06-17 10:05:10 +08:00
commit 6d1dd5d56c
28 changed files with 2167 additions and 0 deletions

3
.gitignore vendored Normal file
View File

@@ -0,0 +1,3 @@
__pycache__/
*.pyc
.helloagents/

70
CLAUDE.md Normal file
View File

@@ -0,0 +1,70 @@
# CLAUDE.md
This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository.
## What this is
A mobile-first, multiplayer drinking-game hub (酒桌派对 / "Drinking Game Hub"). Players open a web page on their phone, enter a nickname, and create or join a 4-character room. The `GAMES` registry declares four games (大话骰 dice, 德州扑克 poker, 六张牌 six, 抓二游 catch), but **only 大话骰 (Liar's Dice, `game="dice"`) has real backend logic** — the other three exist only as lobby cards and static prototype boards.
## Running and testing
No build step, no dependencies — pure Python standard library + vanilla JS/CSS.
```bash
python3 server.py # starts ThreadingHTTPServer on 0.0.0.0:8765
```
Then open `http://127.0.0.1:8765/live.html` (the real MVP). The server seeds one demo room on startup.
Tests are standalone scripts, run individually:
```bash
python3 test_flow.py # HTTP test — requires server.py already running on :8765
python3 test_multi_play.py # in-process test — imports server.py directly
```
Two test styles, distinguishable by their second line:
- **HTTP tests** (`base='http://127.0.0.1:8765'`) drive the live server over the API; start `server.py` first.
- **In-process tests** (`importlib.util.spec_from_file_location(...)`) import `server.py` and call functions directly with no HTTP. **Gotcha:** these hardcode the path `/var/minis/workspace/drinking-games-ui/server.py`. That path predates this checkout (`/home/xiaoyu/projects/drinking-games-ui`); update the literal before running in-process tests here.
Tests assert by printing — there is no test runner or assertion framework. Read the printed output to judge pass/fail.
## Architecture
**`server.py`** — the entire backend in one file. It is both the API server and the static file server (`SimpleHTTPRequestHandler` rooted at the repo dir). All state lives in the module-level `rooms` dict; **nothing is persisted** — restarting the server wipes every room. A room is a plain dict (see `create_room`) holding players, `status` (`waiting`/`playing`/`result`), `currentCall`, `pendingAction`, `lastResult`, `history`, and `roundCalls`.
The API is a flat list of POST endpoints in `Handler.do_POST`, plus GET `/api/rooms` and `/api/room`. Every mutation function returns the room, and the handler wraps it with `public_room(room, viewer_id)` before sending. `public_room` is the **only** serialization boundary — it controls per-viewer visibility via `visible_dice` (you see your own dice while playing; everyone's dice only on reveal). When adding fields the client needs, add them in `public_room`, not just on the room dict.
Errors are signaled by raising `ValueError` with a user-facing Chinese message; `do_POST` catches all exceptions and returns `{"ok": False, "error": str(e)}` with HTTP 400. Follow this pattern — validate and raise `ValueError`, don't return error dicts from game functions.
**Frontends — two separate, unrelated UIs:**
- `live.html` is the **real** app: a single self-contained file (inline `<script>` + `<style>`) wired to the backend API. This is what `server.py` points users to.
- `index.html` + `app.js` + `style.css` is an **older static prototype** with hardcoded mock data and no network calls. Don't confuse the two; changes to gameplay belong in `live.html` and `server.py`.
`style.css` is shared by both. `live.html` adds most of its own styling inline.
## Realtime model (WebSocket)
`live.html` runs the live room over a single WebSocket to `/ws`. The WebSocket server is implemented in **pure stdlib** (RFC6455 handshake + framing — `ws_handshake`/`ws_read_frame`/`ws_build_frame` in `server.py`), so the project keeps its zero-dependency, `python3 server.py`-and-go property. The handshake is detected at the top of `Handler.do_GET`; once upgraded, the request thread hijacks the raw socket and runs `ws_serve` until the socket closes.
Client→server messages are `{type, ...}` (`create`/`join`/`ready`/`start`/`call`/`open`/`split`/`steal`/`multi`/`respond`/`respondMulti`/`settings`/`lock`/`kick`/`addBot`/`leave`/`rooms`), dispatched by `handle_ws`, which calls the same game functions the REST layer uses. After every mutation the server **broadcasts** `public_room` (per-viewer dice visibility) to all connections in that room via `broadcast_room`, plus a lobby `rooms` list to lobby connections. The client (`live.html` `send`/`handleWS`) fires an action and waits for the pushed `room` message — it does not optimistically re-render.
**Bots are server-driven.** After any gameplay-advancing action, `handle_ws` calls `schedule_bots`, which uses a `threading.Timer` (~0.9s) to run `run_bots``bot_turn` (one decision per step), broadcasts, and re-schedules until it's a human's turn or the round ends. There is no client bot loop. Bot decisions use `random` (call vs. open, accept vs. counter).
**Concurrency:** `state_lock` (an `RLock`) guards `rooms` + `ws_clients` mutations across request threads and bot-timer threads; each connection has its own `send_lock`. The server runs with `daemon_threads=True` so long-lived WS threads don't block shutdown.
**REST is retained** (`do_POST` / `GET /api/room`/`/api/rooms`) for the HTTP test scripts and as a fallback, and is *not* lock-guarded — REST and WS are not meant to drive the same room concurrently. `PROJECT_DESIGN.md` predates the WebSocket work; treat it as historical intent.
Note: some HTTP test scripts (`test_flow.py`, `test_multiplayer.py`) are stale and fail at `start` against current rules (they start with <2 players or an unready guest — both rejected by `start_room`); this is a test-data issue, not a server regression.
## 大话骰 rule engine (the core complexity)
The dice rules are non-obvious and spread across several small functions in `server.py`. Key invariants, enforced in `call_dice` / `is_valid_raise` / `call_mode` / `point_rank`:
- **斋 (zhai) vs 飞 (fly)** are the two call modes. In 斋 mode the point counts literally; in 飞 mode all `1`s are wild and count toward any point (except point `1` itself). `count_actual` computes the real total for a call.
- **N-player floor:** minimum opening call is `min_count(room)` = N "ones" (`min_count` = player count). Max is `max_count` = N×5 (five dice each).
- **Point ordering in 斋:** `1` is the **highest** point, so the order is `2 < 3 < 4 < 5 < 6 < 1` (`point_rank` maps `1``7`). Calling a literal `1` is always 斋, never 飞.
- **Breaking 斋 into 飞 ("破斋") requires doubling** the count (`count >= old.count * 2`). This is checked explicitly in `call_dice` with a dedicated error.
- **Challenges** (`pendingAction`): `open_dice` (开骰), `split_dice` (劈), `steal_split` (抢开), and the multi-target `start_multi_direct` (开/劈 two or three players). These set a `pendingAction` with a `stage` state machine resolved by `respond_challenge` (single target) or `respond_multi` (two-target). Stakes escalate through accept / decline / counter (反劈). Read the stage transitions before touching this — `wait_caller_counter``wait_opener_accept` etc.
When changing rules, update the validation in `call_dice`, the predicted-next-call helper `next_call`, the AI's choices in `bot_turn`, **and** the `diceRule` object inside `public_room` (the client renders hints from it) together — they encode the same rules in different forms and drift silently if edited in isolation.

176
PROJECT_DESIGN.md Normal file
View File

@@ -0,0 +1,176 @@
# 喝酒游戏集合项目设计方案
## 项目定位
项目名称暂定:**酒桌派对 Drinking Game Hub**
这是一个适合线下聚会、KTV、酒吧、朋友酒局使用的多人喝酒游戏集合。用户只需要手机打开网页输入昵称即可创建或加入房间。
## 游戏集合
首期包含 4 个游戏:
1. **德州扑克**
- 适合 2-9 人
- 可做筹码、底池、翻牌、转牌、河牌
- 可加入喝酒惩罚例如输家喝、All-in 输家翻倍
2. **大话骰**
- 适合 2-12 人
- 支持叫点、开骰、斋/飞/劈等扩展规则
- 酒局热场核心玩法
3. **六张牌**
- 适合 2-8 人
- 每人六张牌,快速比牌
- 节奏快,适合短局惩罚
4. **抓二游**
- 适合 4-8 人
- 抓牌、组队、对抗、结算
- 可做地方玩法扩展
## 核心用户流程
```text
输入昵称 → 进入大厅 → 选择游戏 → 创建/加入房间 → 等待玩家 → 房主开始 → 游戏进行 → 结算喝酒惩罚 → 再来一局
```
## 页面结构
### 1. 首页 / 大厅
功能:
- 输入昵称
- 快速开房
- 输入房间号加入
- 随机加入
- 游戏列表
- 可加入房间列表
### 2. 游戏选择页
每个游戏卡片展示:
- 游戏名称
- 图标
- 人数范围
- 游戏标签
- 简短玩法说明
- 创建房间按钮
### 3. 房间页
功能:
- 房间号
- 玩家列表
- 房主标识
- 准备状态
- 暂停加入
- 邀请链接
- 开始游戏
- 退出房间
### 4. 游戏页
不同游戏独立布局,但保留统一顶部:
- 当前房间
- 当前玩家
- 轮次状态
- 操作按钮
- 聊天/提示区
- 喝酒惩罚提示
### 5. 结算页
展示:
- 本局赢家
- 本局输家
- 喝酒惩罚
- 再来一局
- 返回大厅
## 视觉风格
建议风格:
- 深色背景
- 霓虹酒吧感
- 橙色、金色、玫红点缀
- 卡片式布局
- 移动端优先
- 按钮大、操作简单
关键词:
```text
酒吧感、派对感、年轻、刺激、手机友好、无需登录、快速开局
```
## 技术建议
### 免费轻量版
适合快速上线:
- 前端HTML / CSS / JavaScript 或 React
- 后端Flask / FastAPI / Node.js
- 实时通信WebSocket
- 数据存储:内存 + SQLite
- 部署1Panel / Docker / Nginx
### 推荐正式版
- 前端React + Tailwind CSS + shadcn/ui
- 后端FastAPI 或 Flask
- 实时通信Socket.IO / WebSocket
- 数据库SQLite 起步,后续 PostgreSQL
- 房间状态Redis 可选
## MVP 优先级
### 第一阶段:大厅 + 房间系统
必须做:
- 昵称进入
- 游戏选择
- 创建房间
- 加入房间
- 玩家列表
- 房主开始
- 暂停加入
### 第二阶段:先做大话骰
原因:
- 最适合喝酒
- 规则容易
- UI 互动强
- 最容易出氛围
### 第三阶段:加入德州扑克
加入:
- 发牌
- 公共牌
- 筹码
- 回合操作
- 胜负结算
### 第四阶段:六张牌、抓二游
作为扩展玩法加入。
## 当前原型文件
- `index.html`:大厅 UI 原型
- `style.css`:视觉样式
可以直接在 Minis 里预览。

23
app.js Normal file
View File

@@ -0,0 +1,23 @@
const games = {
poker:{name:'德州扑克',icon:'♠️',people:'2-9人',tag:'策略',accent:'#ff4d6d',desc:'经典翻牌局,支持筹码、公共牌、喝酒惩罚。'},
dice:{name:'大话骰',icon:'🎲',people:'2-12人',tag:'酒局',accent:'#ffd166',desc:'叫点、开骰、劈/斋玩法,最适合酒桌热场。'},
six:{name:'六张牌',icon:'🃏',people:'2-8人',tag:'快节奏',accent:'#38bdf8',desc:'每人六张牌,快速比牌,节奏快、惩罚明确。'},
catch:{name:'抓二游',icon:'🧩',people:'4-8人',tag:'组队',accent:'#43e97b',desc:'抓牌、组队、对抗、结算,适合地方酒桌玩法。'}
};
let state={game:'dice',room:'A102',locked:false,history:['home']};
const $=s=>document.querySelector(s);const $$=s=>document.querySelectorAll(s);
const players=['小鱼','阿杰','Mia','老陈','可乐','Luna'];
function toast(t){const el=$('#toast');el.textContent=t;el.classList.add('show');setTimeout(()=>el.classList.remove('show'),1200)}
function show(id,push=true){$$('.screen').forEach(x=>x.classList.remove('active'));$('#'+id).classList.add('active');$$('.bottom-nav a').forEach(a=>a.classList.toggle('active',a.dataset.go===id));$('[data-back]').hidden=id==='home';$('#subtitle').textContent=id==='home'?'Drinking Game Hub':id==='room'?'房间等待中':id==='game'?'游戏进行中':id==='result'?'本局结算':'个人中心';if(push&&state.history.at(-1)!==id)state.history.push(id);window.scrollTo(0,0)}
function openModal(type){let html='';if(type==='create')html=`<p class="eyebrow">Create Room</p><h2>创建房间</h2><p class="lead">选择一个游戏,创建后邀请朋友加入。</p><div class="modal-actions">${Object.entries(games).map(([k,g])=>`<button data-create="${k}">${g.icon} ${g.name}</button>`).join('')}</div>`;if(type==='join')html=`<p class="eyebrow">Join Room</p><h2>输入房间号</h2><p class="lead">输入朋友分享的 4 位房间号。</p><input id="joinCode" value="A102" /><div class="modal-actions"><button data-join>加入房间</button></div>`;if(type==='rules')html=`<p class="eyebrow">Rules</p><h2>通用喝酒规则</h2><p class="lead">1. 房主创建房间并选择游戏。<br>2. 所有人准备后开始。<br>3. 每局结束按惩罚喝酒。<br>4. 可设置输家 1 杯、2 杯或自定义。<br>5. 理性饮酒,禁止劝酒。</p>`;$('#modalContent').innerHTML=html;$('#modal').classList.add('show')}
function closeModal(){$('#modal').classList.remove('show')}
function enterRoom(game='dice'){state.game=game;state.room=Math.random().toString(36).slice(2,6).toUpperCase();renderRoom();closeModal();show('room')}
function renderHome(){const grid=$('#gameGrid');grid.innerHTML=Object.entries(games).map(([k,g])=>`<article class="game-card" style="--accent:${g.accent}" data-create="${k}"><div class="game-icon">${g.icon}</div><div><h3>${g.name}</h3><p>${g.desc}</p></div><div class="meta"><span>${g.people}</span><span>${g.tag}</span></div></article>`).join('');$('#roomList').innerHTML=[['老友酒局','dice','5/8'],['KTV 第三局','poker','6/9'],['六张牌快局','six','4/6']].map(r=>`<article class="room-card"><div><h3>${r[0]} · ${games[r[1]].name}</h3><p>房主:阿杰 · ${r[2]} 人 · 等待中</p></div><button data-room="${r[1]}">加入</button></article>`).join('')}
function renderRoom(){const g=games[state.game];$('#roomCode').textContent=state.room;$('#roomTitle').textContent=g.name+'房间';$('#roomDesc').textContent=g.desc+' 等待玩家加入,房主可调整规则并开始游戏。';$('#playerCount').textContent=players.length+'/'+g.people.split('-').pop();$('#players').innerHTML=players.map((p,i)=>`<div class="player"><div class="avatar">${i===0?'👑':'🙂'}</div><strong>${p}</strong><span>${i===0?'房主':'已准备'}</span></div>`).join('')}
function startGame(){const g=games[state.game];$('#gameTitle').textContent=g.name;$('#gameMode').textContent=state.room+' · Round 03';if(state.game==='dice'){$('#gameHint').textContent='轮到阿杰叫点,小鱼可以选择跟、加码或开骰。';$('#gameBoard').innerHTML=`<div class="board"><div class="call-box"><p>当前叫点</p><strong>5 个 4</strong><p class="lead">上一手:阿杰</p></div><div class="dice-row"><div class="die">⚀</div><div class="die">⚂</div><div class="die">⚃</div><div class="die">⚅</div><div class="die">⚁</div></div></div>`;$('#gameActions').innerHTML=`<button>加码</button><button data-action="finish">开骰</button>`}
else if(state.game==='poker'){$('#gameHint').textContent='翻牌圈,当前底池 18 杯,轮到小鱼操作。';$('#gameBoard').innerHTML=`<div class="board"><p class="eyebrow">公共牌</p><div class="card-row"><div class="play-card red">A♥</div><div class="play-card">K♠</div><div class="play-card red">9♦</div></div><p class="eyebrow">你的手牌</p><div class="card-row"><div class="play-card">Q♣</div><div class="play-card red">Q♥</div></div></div>`;$('#gameActions').innerHTML=`<button>跟注</button><button>加注</button><button class="ghost-btn">弃牌</button><button data-action="finish">结算</button>`}
else if(state.game==='six'){$('#gameHint').textContent='六张牌亮牌阶段,比较最大组合。';$('#gameBoard').innerHTML=`<div class="board"><p class="eyebrow">你的六张牌</p><div class="card-row"><div class="play-card red">A♥</div><div class="play-card">A♠</div><div class="play-card red">10♦</div><div class="play-card">8♣</div><div class="play-card red">6♥</div><div class="play-card">3♠</div></div></div>`;$('#gameActions').innerHTML=`<button>换一张</button><button data-action="finish">亮牌</button>`}
else {$('#gameHint').textContent='抓二游组队完成,开始出牌对抗。';$('#gameBoard').innerHTML=`<div class="board"><div class="call-box"><p>当前队伍</p><strong>小鱼 + Mia</strong><p class="lead">对阵 阿杰 + 老陈</p></div><div class="card-row"><div class="play-card red">2♥</div><div class="play-card">J♠</div><div class="play-card red">7♦</div><div class="play-card">5♣</div></div></div>`;$('#gameActions').innerHTML=`<button>出牌</button><button>不要</button><button data-action="finish">结算</button>`}show('game')}
function finish(){const loser=['阿杰','老陈','Mia','可乐'][Math.floor(Math.random()*4)];$('#resultTitle').textContent=$('#nickname').value||'小鱼'+'获胜';$('#resultDesc').textContent=`${loser} 本局失败,按规则喝 1 杯。`;$('#punishText').textContent=`${loser} 喝 1 杯`;show('result')}
document.addEventListener('click',e=>{const t=e.target.closest('[data-open],[data-close],[data-create],[data-room],[data-go],[data-action],[data-back],[data-join]');if(!t)return;if(t.dataset.open)openModal(t.dataset.open);if(t.dataset.close!==undefined)closeModal();if(t.dataset.create)enterRoom(t.dataset.create);if(t.dataset.room){state.game=t.dataset.room;renderRoom();show('room')}if(t.dataset.go){if(t.dataset.go==='room')renderRoom();show(t.dataset.go)}if(t.dataset.back!==undefined){state.history.pop();show(state.history.pop()||'home')}if(t.dataset.join!==undefined){state.room=$('#joinCode').value||'A102';renderRoom();closeModal();show('room')}if(t.dataset.action==='enter')toast('欢迎进入大厅');if(t.dataset.action==='randomJoin'){state.game='dice';renderRoom();show('room')}if(t.dataset.action==='startGame')startGame();if(t.dataset.action==='finish')finish();if(t.dataset.action==='again')startGame();if(t.dataset.action==='copyInvite')toast('邀请链接已复制');if(t.dataset.action==='toggleLock'){state.locked=!state.locked;toast(state.locked?'已暂停加入':'已开放加入')}if(t.dataset.action==='toast')toast('房间列表已刷新')});
renderHome();renderRoom();

View File

@@ -0,0 +1,15 @@
[Unit]
Description=Drinking Games Hub (大话骰)
After=network.target
[Service]
Type=simple
WorkingDirectory=/opt/drinking-games
ExecStart=/usr/bin/python3 /opt/drinking-games/server.py
Restart=always
RestartSec=2
User=root
# server.py 监听 0.0.0.0:8765由 Nginx 反代;防火墙应只放行 80/443
[Install]
WantedBy=multi-user.target

View File

@@ -0,0 +1,35 @@
# Nginx 反向代理 — dn.akqp.online → 127.0.0.1:8765 (server.py)
# 支持 WebSocket/ws。HTTPS 由 certbot --nginx 自动追加 443 段与跳转。
map $http_upgrade $connection_upgrade {
default upgrade;
'' close;
}
server {
listen 80;
listen [::]:80;
server_name dn.akqp.online;
# 入口直接打开真实 MVP 页面
location = / {
return 302 /live.html;
}
location / {
proxy_pass http://127.0.0.1:8765;
proxy_http_version 1.1;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
# WebSocket 升级(/ws 走这里)
proxy_set_header Upgrade $http_upgrade;
proxy_set_header Connection $connection_upgrade;
# 长连接保活,避免 WS 被默认 60s 读超时切断
proxy_read_timeout 3600s;
proxy_send_timeout 3600s;
}
}

90
deploy/setup.sh Executable file
View File

@@ -0,0 +1,90 @@
#!/usr/bin/env bash
#
# 在服务器上一键部署「酒桌派对 / 大话骰」
# - 把项目复制到 /opt/drinking-games
# - systemd 常驻运行 server.py (127.0.0.1 视角的 8765 端口,外部由 Nginx 反代)
# - Nginx 反向代理 dn.akqp.online -> 8765含 WebSocket
# - certbot 自动签发 Let's Encrypt 证书并配置 HTTPS + 80→443 跳转
#
# 用法(在已 scp 上来的项目根目录里执行):
# sudo bash deploy/setup.sh
#
set -euo pipefail
DOMAIN="dn.akqp.online"
APP_DIR="/opt/drinking-games"
EMAIL="PumpkinPiekur@hairdresser.net" # certbot 续期通知邮箱
SRC_DIR="$(cd "$(dirname "$0")/.." && pwd)"
log() { printf '\n\033[1;36m== %s\033[0m\n' "$*"; }
die() { printf '\n\033[1;31m[ERR] %s\033[0m\n' "$*" >&2; exit 1; }
[ "$(id -u)" = "0" ] || die "请用 root 执行sudo bash deploy/setup.sh"
command -v python3 >/dev/null || die "未找到 python3"
# ---------- 0. Cloudflare 代理检测 ----------
log "检测域名解析Let's Encrypt 需直连源站)"
SERVER_IP="$(curl -fsS --max-time 10 https://api.ipify.org || echo '?')"
RESOLVED="$(getent ahostsv4 "$DOMAIN" | awk '{print $1}' | sort -u | tr '\n' ' ')"
echo " 本机公网 IP : $SERVER_IP"
echo " 域名解析到 : ${RESOLVED:-(无)}"
if [ "$SERVER_IP" != "?" ] && ! echo " $RESOLVED " | grep -q " $SERVER_IP "; then
cat <<EOF
⚠ $DOMAIN 当前未直接解析到本机(可能在 Cloudflare 橙云代理后面)。
Let's Encrypt 的 HTTP-01 验证可能失败。建议二选一:
a) 在 Cloudflare 把该记录临时切到 "DNS only"(灰云)直连本机,签完证书再开回代理;
开回代理后请把 SSL/TLS 模式设为 Full (strict)。
b) 保持代理,但确保 SSL 模式为 Full 且未强制把 80 端口跳转。
脚本将继续,签证书若失败会停在 certbot 步骤,按上面处理后重跑即可。
EOF
fi
# ---------- 1. 安装依赖 ----------
log "安装 Nginx 与 certbot"
export DEBIAN_FRONTEND=noninteractive
apt-get update -y
apt-get install -y nginx certbot python3-certbot-nginx rsync
# ---------- 2. 部署应用 ----------
log "复制项目到 $APP_DIR"
mkdir -p "$APP_DIR"
rsync -a --delete \
--exclude '.git' --exclude '__pycache__' --exclude 'deploy' \
--exclude 'test_*.py' \
"$SRC_DIR"/ "$APP_DIR"/
log "安装 systemd 服务 drinking-games"
install -m 644 "$SRC_DIR/deploy/drinking-games.service" /etc/systemd/system/drinking-games.service
systemctl daemon-reload
systemctl enable drinking-games
systemctl restart drinking-games
sleep 1
systemctl is-active --quiet drinking-games || { journalctl -u drinking-games --no-pager -n 30; die "服务启动失败"; }
curl -fsS --max-time 5 http://127.0.0.1:8765/api/rooms >/dev/null && echo " 后端本地自检 OK" || echo " ⚠ 后端本地自检失败,稍后检查日志"
# ---------- 3. Nginx 反代 ----------
log "配置 Nginx 反向代理"
install -m 644 "$SRC_DIR/deploy/nginx-dn.akqp.online.conf" /etc/nginx/sites-available/dn.akqp.online.conf
ln -sf /etc/nginx/sites-available/dn.akqp.online.conf /etc/nginx/sites-enabled/dn.akqp.online.conf
[ -e /etc/nginx/sites-enabled/default ] && rm -f /etc/nginx/sites-enabled/default || true
nginx -t
systemctl reload nginx
# ---------- 4. 防火墙(若启用了 ufw ----------
if command -v ufw >/dev/null && ufw status | grep -q "Status: active"; then
log "放行 80/443关闭 8765 外部访问"
ufw allow 'Nginx Full' || true
ufw deny 8765/tcp || true
fi
# ---------- 5. HTTPS 证书 ----------
log "签发 Let's Encrypt 证书并启用 HTTPS"
certbot --nginx -d "$DOMAIN" \
--non-interactive --agree-tos -m "$EMAIL" \
--redirect || die "证书签发失败(多为 Cloudflare 代理导致 HTTP-01 验证不通),按开头提示处理后重跑本脚本"
systemctl reload nginx
log "完成"
echo " 访问: https://$DOMAIN/ (自动跳到 /live.html"
echo " 服务: systemctl status drinking-games | journalctl -u drinking-games -f"
echo " 证书自动续期由 certbot.timer 负责,可用 'certbot renew --dry-run' 验证。"

72
index.html Normal file
View File

@@ -0,0 +1,72 @@
<!doctype html>
<html lang="zh-CN">
<head>
<meta charset="utf-8" />
<meta name="viewport" content="width=device-width, initial-scale=1" />
<title>酒桌派对 - 可点击原型</title>
<link rel="stylesheet" href="style.css?v=4" />
</head>
<body>
<main class="app-shell">
<header class="app-top">
<button class="icon-btn" data-back hidden></button>
<div class="brand"><span class="logo">🍻</span><div><strong>酒桌派对</strong><small id="subtitle">Drinking Game Hub</small></div></div>
<button class="ghost-btn small" data-open="rules">规则</button>
</header>
<section class="screen active" id="home">
<div class="hero-card">
<p class="eyebrow">多人聚会 · 手机即开 · 房间制</p>
<h1>一站式喝酒游戏大厅</h1>
<p class="lead">集合德州扑克、大话骰、六张牌、抓二游适合线下酒局、朋友聚会、KTV、酒吧局。</p>
<div class="join-panel">
<label>你的昵称</label>
<div class="input-row"><input id="nickname" placeholder="例如:小鱼" value="小鱼" /><button data-action="enter">进入大厅</button></div>
</div>
</div>
<div class="quick-actions">
<button class="primary-action" data-open="create">快速开房</button>
<button data-open="join">输入房间号</button>
<button data-action="randomJoin">随机加入</button>
</div>
<div class="section-head"><div><p class="eyebrow">Game Collection</p><h2>选择游戏</h2></div><span class="online-pill">128 人在线</span></div>
<section class="game-grid" id="gameGrid"></section>
<section class="room-section"><div class="section-head compact"><div><p class="eyebrow">Open Rooms</p><h2>可加入房间</h2></div><button class="ghost-btn dark" data-action="toast">刷新</button></div><div class="room-list" id="roomList"></div></section>
</section>
<section class="screen" id="room">
<div class="room-hero">
<p class="eyebrow">Room <span id="roomCode">A102</span></p>
<h1 id="roomTitle">大话骰房间</h1>
<p class="lead" id="roomDesc">等待玩家加入,房主可调整规则并开始游戏。</p>
<div class="room-actions"><button data-action="copyInvite">复制邀请</button><button class="ghost-btn" data-action="toggleLock">暂停加入</button></div>
</div>
<div class="panel"><div class="section-head compact"><h2>玩家</h2><span class="online-pill" id="playerCount">5/8</span></div><div class="players" id="players"></div></div>
<div class="panel"><div class="section-head compact"><h2>房间设置</h2></div><div class="settings"><label>喝酒惩罚<select><option>输家 1 杯</option><option>输家 2 杯</option><option>自定义</option></select></label><label>游戏节奏<select><option>普通</option><option>快速</option><option>疯狂</option></select></label></div></div>
<button class="wide primary-action" data-action="startGame">开始游戏</button>
</section>
<section class="screen" id="game">
<div class="game-header"><p class="eyebrow" id="gameMode">Dice Round 03</p><h1 id="gameTitle">大话骰</h1><p class="lead" id="gameHint">轮到阿杰叫点。</p></div>
<div id="gameBoard"></div>
<div class="action-dock" id="gameActions"></div>
</section>
<section class="screen" id="result">
<div class="result-card"><div class="trophy">🏆</div><p class="eyebrow">Round Result</p><h1 id="resultTitle">小鱼获胜</h1><p class="lead" id="resultDesc">阿杰本局失败,按规则喝 1 杯。</p><div class="punish">🍺 <strong id="punishText">阿杰喝 1 杯</strong></div><button class="wide primary-action" data-action="again">再来一局</button><button class="wide ghost-btn" data-go="home">返回大厅</button></div>
</section>
<section class="screen" id="profile">
<div class="panel"><p class="eyebrow">My Stats</p><h1>我的酒桌战绩</h1><div class="stats"><div><strong>12</strong><span>胜场</span></div><div><strong>7</strong><span>喝酒</span></div><div><strong>4</strong><span>开房</span></div></div></div>
</section>
<nav class="bottom-nav"><a class="active" data-go="home">大厅</a><a data-go="room">房间</a><a data-go="result">结算</a><a data-go="profile">我的</a></nav>
</main>
<div class="modal" id="modal">
<div class="modal-card"><button class="close" data-close>×</button><div id="modalContent"></div></div>
</div>
<div class="toast" id="toast">已刷新</div>
<script src="app.js?v=2"></script>
</body>
</html>

291
live.html Normal file

File diff suppressed because one or more lines are too long

787
server.py Normal file
View File

@@ -0,0 +1,787 @@
#!/usr/bin/env python3
import json, random, string, time, hashlib, base64, struct, threading
from http.server import ThreadingHTTPServer, SimpleHTTPRequestHandler
from pathlib import Path
from urllib.parse import urlparse, parse_qs
ROOT = Path(__file__).resolve().parent
GAMES = {"dice":{"name":"大话骰","max":12},"poker":{"name":"德州扑克","max":9},"six":{"name":"六张牌","max":8},"catch":{"name":"抓二游","max":8}}
rooms = {}
BOT_NAMES = ["阿杰","Mia","老陈","可乐","Luna","大熊","小白","阿豪"]
def now_ms(): return int(time.time()*1000)
def new_id(prefix="p"): return prefix + "".join(random.choice(string.ascii_lowercase+string.digits) for _ in range(10))
def room_code():
while True:
code="".join(random.choice(string.ascii_uppercase+string.digits) for _ in range(4))
if code not in rooms: return code
def make_player(name, host=False, player_id=None, bot=False, ready=None):
if ready is None: ready = True if (host or bot) else False
return {"id":player_id or new_id("b" if bot else "p"),"name":(name or "玩家")[:16],"host":host,"bot":bot,"ready":ready,"cups":0,"dice":[],"stats":{"splits":0,"counters":0,"declines":0,"cupsDrunk":0}}
def find_player(room, player_id): return next((p for p in room["players"] if p["id"]==player_id), None)
def require_player(room, player_id):
p=find_player(room, player_id)
if not p: raise ValueError("当前身份不在房间内")
return p
def require_host(room, player_id):
p=require_player(room, player_id)
if not p.get("host"): raise ValueError("只有房主可以操作")
return p
def current_player(room):
if not room["players"]: return None
room["turnIndex"]%=len(room["players"])
return room["players"][room["turnIndex"]]
def min_count(room): return max(1,len(room["players"]))
def max_count(room): return len(room["players"])*5
def call_mode(room,count,old=None,requested=None,point=None):
count = int(count)
if requested in ["",""]:
if requested == "" and int(point or 0) == 1:
return "" # 先保留请求后续校验会明确拒绝叫1不能飞
return requested
if int(point or 0) == 1:
return "" # 未显式选择时,叫 1 默认只能是斋
if not old:
# 仅首叫按人数阈值判断N、N+1 为斋N+2 起飞
return "" if count >= min_count(room)+2 else ""
if old.get("mode") == "":
# 默认继续飞;若要平斋,由前端显式传 mode=斋
return ""
# 斋局后续:未达到上一手数量 2 倍时仍是斋;达到 2 倍才破斋飞
return "" if count >= int(old["count"])*2 else ""
def point_rank(mode, point):
point = int(point)
if mode == "":
return 7 if point == 1 else point # 斋局里 1 最大2<3<4<5<6<1
return point
def count_actual(room,call):
point=int(call["point"]); flying=call.get("mode")==""
return sum(1 for p in room["players"] for d in p["dice"] if d==point or (flying and point!=1 and d==1))
def is_valid_raise(room, old, count, point, mode):
if not old: return True
old_mode = old.get("mode", "")
if old_mode == "":
if mode == "":
return count >= old["count"] * 2
# 继续斋:数量可增加;同数量时按斋点数顺序 2<3<4<5<6<1
if count > old["count"]: return True
if count == old["count"]: return point_rank("", point) > point_rank("", old["point"])
return False
# 上一手已飞
if mode == "":
# 飞后切斋/平斋:不能降数量;同数量或更高数量都可进入斋
return count >= old["count"]
# 继续飞:数量增加,或同数量更大点数
if count > old["count"]: return True
if count == old["count"]: return int(point) > int(old["point"])
return False
def next_call(room,old):
if not old: return {"count":min_count(room),"point":1}
if old.get("mode") == "":
order=[2,3,4,5,6,1]
idx=order.index(old["point"]) if old["point"] in order else 0
if idx < len(order)-1:
return {"count":old["count"],"point":order[idx+1]}
return {"count":old["count"]+1,"point":2}
if old["point"]<6: return {"count":old["count"],"point":old["point"]+1}
return {"count":old["count"]+1,"point":1}
def recent_call_targets(room, count):
seen=set(); items=[]
for entry in reversed(room.get("roundCalls", [])):
pid=entry["playerId"]
if pid in seen: continue
seen.add(pid)
items.append(entry)
if len(items)>=count: break
return items
def add_drink(player, cups):
player["cups"] += cups
player.setdefault("stats",{}).setdefault("cupsDrunk",0)
player["stats"]["cupsDrunk"] += cups
def visible_dice(room, player, viewer_id):
if room["status"]=="result" and room.get("lastResult",{}).get("revealDice", True): return player["dice"]
if room["status"]=="playing" and viewer_id and player["id"]==viewer_id: return player["dice"]
return []
def public_room(room, viewer_id=None):
cur=current_player(room)
return {"code":room["code"],"game":room["game"],"gameName":GAMES[room["game"]]["name"],"maxPlayers":GAMES[room["game"]]["max"],"status":room["status"],"locked":room["locked"],"round":room["round"],"currentCall":room["currentCall"],"turnIndex":room["turnIndex"],"turnPlayerId":cur["id"] if cur else None,"turnPlayerName":cur["name"] if cur else None,"turnPlayerBot":bool(cur and cur.get("bot")),"pendingAction":room.get("pendingAction"),"lastResult":room["lastResult"],"updatedAt":room["updatedAt"],"history":room.get("history",[])[-16:],"settings":room.get("settings",{"cupsPerLoss":1}),"diceRule":{"minCount":min_count(room),"zhaiCounts":[min_count(room),min_count(room)+1],"flyFrom":min_count(room)+2,"maxCount":max_count(room),"zhaiPointOrder":[2,3,4,5,6,1],"breakZhai":"double"},"players":[{"id":p["id"],"name":p["name"],"host":p["host"],"bot":p["bot"],"ready":p["ready"],"cups":p["cups"],"stats":p.get("stats",{}),"dice":visible_dice(room,p,viewer_id)} for p in room["players"]]}
def send_json(h,data,status=200):
body=json.dumps(data,ensure_ascii=False).encode(); h.send_response(status); h.send_header("Content-Type","application/json; charset=utf-8"); h.send_header("Content-Length",str(len(body))); h.send_header("Access-Control-Allow-Origin","*"); h.end_headers(); h.wfile.write(body)
def read_json(h):
n=int(h.headers.get("Content-Length","0") or 0)
return json.loads(h.rfile.read(n).decode() or "{}") if n else {}
def append_history(room, typ, text, player="系统"):
room.setdefault("history",[]).append({"type":typ,"player":player,"text":text,"at":now_ms()})
def transfer_host_if_needed(room):
if any(p.get("host") for p in room["players"]): return
human=next((p for p in room["players"] if not p.get("bot")), None)
target=human or (room["players"][0] if room["players"] else None)
if target:
target["host"]=True; target["ready"]=True; append_history(room,"host",f"{target['name']} 成为新房主",target["name"])
def cleanup_room(code):
room=rooms.get(code)
if room and not any(not p.get("bot") for p in room["players"]):
rooms.pop(code,None)
return True
return False
def create_room(game,name,player_id=None):
game=game if game in GAMES else "dice"; code=room_code(); player=make_player(name,True,player_id,ready=True)
rooms[code]={"code":code,"game":game,"status":"waiting","locked":False,"round":0,"turnIndex":0,"currentCall":None,"pendingAction":None,"lastResult":None,"history":[],"roundCalls":[],"settings":{"cupsPerLoss":1},"players":[player],"updatedAt":now_ms()}
append_history(rooms[code],"system",f"{player['name']} 创建房间",player["name"])
return rooms[code],player
def join_room(code,name,player_id=None):
code=(code or "").upper().strip(); room=rooms.get(code)
if not room: raise ValueError("房间不存在")
existed=find_player(room,player_id) if player_id else None
if existed:
existed["name"]=(name or existed["name"])[:16]; room["updatedAt"]=now_ms(); return room,existed,True
same_name=next((p for p in room["players"] if not p["bot"] and p["name"]==(name or "玩家")[:16]),None)
if same_name:
room["updatedAt"]=now_ms(); return room,same_name,True
if room["locked"]: raise ValueError("房间已暂停加入")
if room["status"]!="waiting": raise ValueError("游戏已开始,暂不能加入;原玩家可重连回房")
if len(room["players"])>=GAMES[room["game"]]["max"]: raise ValueError("房间已满")
player=make_player(name,False,player_id,ready=False); room["players"].append(player); append_history(room,"join",f"{player['name']} 加入房间",player["name"]); room["updatedAt"]=now_ms(); return room,player,False
def set_ready(code, player_id, ready=None):
room=rooms.get((code or "").upper())
if not room: raise ValueError("房间不存在")
if room["status"]=="playing": raise ValueError("游戏中不能切换准备状态")
p=require_player(room,player_id)
p["ready"] = (not p["ready"]) if ready is None else bool(ready)
append_history(room,"ready",f"{p['name']} {'已准备' if p['ready'] else '取消准备'}",p["name"])
room["updatedAt"]=now_ms(); return room
def add_bot(code, player_id=None):
room=rooms.get((code or "").upper())
if not room: raise ValueError("房间不存在")
if player_id: require_host(room,player_id)
if room["status"]=="playing": raise ValueError("游戏中不能添加机器人")
if len(room["players"])>=GAMES[room["game"]]["max"]: raise ValueError("房间已满")
used={p["name"] for p in room["players"]}; name=next((n for n in BOT_NAMES if n not in used), "机器人")
bot=make_player(name,bot=True,ready=True); room["players"].append(bot); append_history(room,"bot",f"机器人 {name} 加入房间",name); room["updatedAt"]=now_ms(); return room
def update_settings(code, player_id, cups_per_loss=None):
room=rooms.get((code or "").upper())
if not room: raise ValueError("房间不存在")
require_host(room,player_id)
if room["status"]=="playing": raise ValueError("游戏中不能修改规则")
if cups_per_loss is not None:
cups=int(cups_per_loss)
if cups<1 or cups>5: raise ValueError("输家杯数必须是 1-5")
room.setdefault("settings",{})["cupsPerLoss"]=cups
append_history(room,"settings",f"输家惩罚改为 {cups}")
room["updatedAt"]=now_ms(); return room
def start_room(code, player_id=None):
room=rooms.get((code or "").upper())
if not room: raise ValueError("房间不存在")
if player_id: require_host(room,player_id)
if room["status"]=="playing": raise ValueError("游戏进行中,不能重新开始")
if len(room["players"])<2: raise ValueError("至少需要 2 人,房主可先添加机器人或等待玩家加入")
not_ready=[p["name"] for p in room["players"] if not p.get("ready")]
if not_ready: raise ValueError("还有玩家未准备:"+"".join(not_ready))
room["status"]="playing"; room["round"]+=1; start_id=(room.get("lastResult") or {}).get("loserId")
room["turnIndex"]=next((i for i,p in enumerate(room["players"]) if p["id"]==start_id),0)
room["currentCall"]=None; room["pendingAction"]=None; room["lastResult"]=None; room["history"]=[]; room["roundCalls"]=[]
for p in room["players"]: p["dice"]=[random.randint(1,6) for _ in range(5)]
append_history(room,"start",f"{room['round']} 局开始")
room["updatedAt"]=now_ms(); return room
def validate_turn(room,player_id=None):
p=current_player(room)
if not p: raise ValueError("房间没有玩家")
if player_id and p["id"]!=player_id: raise ValueError(f"还没轮到你,当前轮到 {p['name']}")
return p
def call_dice(code,player_id,count,point,requested_mode=None):
room=rooms.get((code or "").upper())
if not room: raise ValueError("房间不存在")
if room["game"]!="dice": raise ValueError("当前不是大话骰房间")
if room["status"]!="playing": raise ValueError("游戏未开始或已结算,不能叫骰")
actor=validate_turn(room,player_id); count,point=int(count),int(point)
if point<1 or point>6: raise ValueError("点数必须是 1-6")
if count<min_count(room): raise ValueError(f"{len(room['players'])} 人局最低从 {min_count(room)} 个 1 开始叫")
if count>max_count(room): raise ValueError(f"数量不能超过总骰子数 {max_count(room)}")
if requested_mode == "" and point == 1:
raise ValueError("叫 1 只能是斋,不能叫飞")
if not room["currentCall"] and count == min_count(room) and point != 1:
raise ValueError(f"首叫最低数量时只能叫 {min_count(room)} 个 1")
if not room["currentCall"] and count <= min_count(room)+1 and requested_mode == "":
raise ValueError(f"首轮低数量阶段只能叫斋;{len(room['players'])} 人局前两档 {min_count(room)}{min_count(room)+1} 个都不能直接叫飞")
# 新规则校验见下方:斋局 1 最大,破斋必须翻倍
mode=call_mode(room,count,room["currentCall"],requested_mode,point)
if room["currentCall"] and room["currentCall"].get("mode") == "" and mode == "" and count < room["currentCall"]["count"] * 2:
old = room["currentCall"]
raise ValueError(f"破斋需要翻倍:当前 {old['count']}{old['point']} · 斋,至少要叫 {old['count']*2} 个任意点才能飞")
if not is_valid_raise(room, room["currentCall"],count,point,mode): raise ValueError("叫点必须比上一手大;斋局点数顺序为 2<3<4<5<6<1且破斋必须翻倍")
room["currentCall"]={"count":count,"point":point,"mode":mode,"by":actor["name"],"byId":actor["id"]}
room.setdefault("roundCalls",[]).append({"playerId":actor["id"],"playerName":actor["name"],"count":count,"point":point,"mode":mode,"order":len(room.get("roundCalls",[]))+1})
append_history(room,"call",f"{actor['name']}{count}{point} · {mode}",actor["name"])
room["turnIndex"]=(room["turnIndex"]+1)%len(room["players"]); room["updatedAt"]=now_ms(); return room
def finish_no_reveal(room, loser, cups, text):
loser["cups"] += cups
loser.setdefault("stats",{}).setdefault("cupsDrunk",0)
loser["stats"]["cupsDrunk"] += cups
room["status"] = "result"
room["pendingAction"] = None
room["lastResult"] = {"call": room.get("currentCall"), "actual": None, "opener": None, "openerId": None, "loser": loser["name"], "loserId": loser["id"], "cups": cups, "text": text, "revealDice": False}
append_history(room, "decline", text, loser["name"])
room["updatedAt"] = now_ms()
return room
def resolve_open(room, opener, stake_cups, reason="开骰"):
call = room["currentCall"]
actual = count_actual(room, call)
caller = find_player(room, call.get("byId")) or room["players"][0]
loser = opener if actual >= call["count"] else caller
loser["cups"] += stake_cups
loser.setdefault("stats",{}).setdefault("cupsDrunk",0)
loser["stats"]["cupsDrunk"] += stake_cups
room["status"] = "result"
room["pendingAction"] = None
result_text = f"{opener['name']} {reason}{call['mode']}局实际有 {actual}{call['point']}{loser['name']}{stake_cups}"
room["lastResult"] = {"call": call, "actual": actual, "opener": opener["name"], "openerId": opener["id"], "loser": loser["name"], "loserId": loser["id"], "cups": stake_cups, "text": result_text, "revealDice": True}
append_history(room, "open", result_text, opener["name"])
room["updatedAt"] = now_ms()
return room
def open_dice(code,player_id=None):
room=rooms.get((code or "").upper())
if not room: raise ValueError("房间不存在")
if room["game"]!="dice": raise ValueError("当前不是大话骰房间")
if room["status"]!="playing": raise ValueError("游戏未开始或已结算,不能开骰")
if room.get("pendingAction"): raise ValueError("当前有待处理的开骰/劈")
if not room["currentCall"]: raise ValueError("还没有人叫点,不能开骰")
opener=validate_turn(room,player_id)
caller=find_player(room,room["currentCall"].get("byId"))
if not caller: raise ValueError("上一手叫骰玩家不存在")
base=int(room.get("settings",{}).get("cupsPerLoss",1))
room["pendingAction"]={"type":"open","stage":"wait_caller_counter","initiatorId":opener["id"],"initiatorName":opener["name"],"targetId":caller["id"],"targetName":caller["name"],"baseCups":base,"call":room["currentCall"]}
append_history(room,"challenge",f"{opener['name']} 要开骰,等待 {caller['name']} 是否反劈",opener["name"])
room["updatedAt"]=now_ms(); return room
def split_dice(code,player_id=None):
room=rooms.get((code or "").upper())
if not room: raise ValueError("房间不存在")
if room["game"]!="dice": raise ValueError("当前不是大话骰房间")
if room["status"]!="playing": raise ValueError("游戏未开始或已结算,不能劈")
if room.get("pendingAction"): raise ValueError("当前有待处理的开骰/劈")
if not room["currentCall"]: raise ValueError("还没有人叫点,不能劈")
opener=validate_turn(room,player_id)
caller=find_player(room,room["currentCall"].get("byId"))
if not caller: raise ValueError("上一手叫骰玩家不存在")
base=int(room.get("settings",{}).get("cupsPerLoss",1))
opener.setdefault("stats",{}).setdefault("splits",0)
opener["stats"]["splits"] += 1
room["pendingAction"]={"type":"split","stage":"wait_caller_accept","initiatorId":opener["id"],"initiatorName":opener["name"],"targetId":caller["id"],"targetName":caller["name"],"baseCups":base,"call":room["currentCall"]}
append_history(room,"split",f"{opener['name']}{caller['name']},等待受/不受/反劈",opener["name"])
room["updatedAt"]=now_ms(); return room
def respond_challenge(code, player_id, response):
room=rooms.get((code or "").upper())
if not room: raise ValueError("房间不存在")
pending=room.get("pendingAction")
if not pending: raise ValueError("当前没有待处理的开骰/劈")
base=int(pending.get("baseCups",1))
initiator=find_player(room,pending["initiatorId"])
target=find_player(room,pending["targetId"])
if not initiator or not target: raise ValueError("挑战玩家不存在")
stage=pending["stage"]
if stage=="wait_caller_counter":
if player_id!=target["id"]: raise ValueError(f"等待 {target['name']} 是否反劈")
if response=="no_counter": return resolve_open(room,initiator,base,"开骰")
if response=="counter":
target.setdefault("stats",{}).setdefault("counters",0)
target["stats"]["counters"] += 1
pending["stage"]="wait_opener_accept"; pending["counterType"]="open"; pending["declineCups"]=base; pending["acceptCups"]=base+1
append_history(room,"counter",f"{target['name']} 反劈,等待 {initiator['name']} 受/不受",target["name"])
room["updatedAt"]=now_ms(); return room
if stage=="wait_caller_accept":
if player_id!=target["id"]: raise ValueError(f"等待 {target['name']} 受/不受/反劈")
if response=="decline":
target.setdefault("stats",{}).setdefault("declines",0)
target["stats"]["declines"] += 1
return finish_no_reveal(room,target,base,f"{target['name']} 不受,喝 {base} 杯,本局结束")
if response=="accept": return resolve_open(room,initiator,base+1,"劈后开骰")
if response=="counter":
target.setdefault("stats",{}).setdefault("counters",0)
target["stats"]["counters"] += 1
pending["stage"]="wait_opener_accept"; pending["counterType"]="split"; pending["declineCups"]=base+1; pending["acceptCups"]=base+2
append_history(room,"counter",f"{target['name']} 反劈,等待 {initiator['name']} 受/不受",target["name"])
room["updatedAt"]=now_ms(); return room
if stage=="wait_opener_accept":
if player_id!=initiator["id"]: raise ValueError(f"等待 {initiator['name']} 受/不受")
if response=="decline":
initiator.setdefault("stats",{}).setdefault("declines",0)
initiator["stats"]["declines"] += 1
return finish_no_reveal(room,initiator,int(pending["declineCups"]),f"{initiator['name']} 不受,喝 {pending['declineCups']} 杯,本局结束")
if response=="accept": return resolve_open(room,initiator,int(pending["acceptCups"]),"受反劈后开骰")
raise ValueError("无效应答")
def steal_split(code, player_id=None):
room=rooms.get((code or "").upper())
if not room: raise ValueError("房间不存在")
if room["game"]!="dice": raise ValueError("当前不是大话骰房间")
if room["status"]!="playing": raise ValueError("游戏未开始或已结算,不能抢开")
if room.get("pendingAction"): raise ValueError("当前有待处理的开骰/劈")
if not room["currentCall"]: raise ValueError("还没有人叫点,不能抢开")
actor=require_player(room,player_id); caller=find_player(room,room["currentCall"].get("byId")); nxt=current_player(room)
if not caller: raise ValueError("上一手叫骰玩家不存在")
if actor["id"]==caller["id"]: raise ValueError("叫骰本人不能抢开自己")
if nxt and actor["id"]==nxt["id"]: raise ValueError("下家可直接操作,不能算抢开")
base=int(room.get("settings",{}).get("cupsPerLoss",1))
actor.setdefault("stats",{}).setdefault("splits",0)
actor["stats"]["splits"] += 1
room["pendingAction"]={"type":"steal_split","stage":"wait_caller_accept","initiatorId":actor["id"],"initiatorName":actor["name"],"targetId":caller["id"],"targetName":caller["name"],"baseCups":base,"call":room["currentCall"]}
append_history(room,"split",f"{actor['name']} 抢开 {caller['name']},等待受/不受/反劈",actor["name"])
room["updatedAt"]=now_ms(); return room
def multi_outcome(initiator, target_names, drink_map):
"""统一多家结算展示数据winners=未喝酒的参与者drinks=各人喝杯明细。"""
participants=[initiator['name']]+list(target_names)
winners=[n for n in participants if drink_map.get(n,0)<=0]
drinks=[{"name":n,"cups":c} for n,c in drink_map.items() if c>0]
return winners, drinks
def multi_calls(room, targets):
"""两家结算各参与者的叫点明细:已开骰的带实际数与是否成立,不受的实际数留空。"""
out=[]
for t in targets:
call=t.get('call',{}) or {}
resolved=t.get('status')=='resolve'
actual=count_actual(room,call) if resolved and call else None
win=(actual>=int(call['count'])) if resolved and call else None
out.append({"name":t['playerName'],"count":call.get('count'),"point":call.get('point'),"mode":call.get('mode'),"actual":actual,"win":win})
return out
def finalize_multi_direct(room, initiator, targets, base, label):
outcomes=[]
for t in targets:
actual=count_actual(room,t)
win = actual >= t["count"]
outcomes.append({"target":t,"actual":actual,"win":win})
winners=[o for o in outcomes if o["win"]]
losers=[o for o in outcomes if not o["win"]]
drink_map={}
def _drink(player, cups): add_drink(player, cups); drink_map[player['name']]=drink_map.get(player['name'],0)+cups
if losers and not winners:
per=base*2
for o in losers: _drink(find_player(room,o["target"]["playerId"]), per)
top=losers[0]["target"]
text=f"{initiator['name']} {label}:三家全输,三家各喝 {per}"
room["lastResult"]={"multi":True,"targets":[{"name":o['target']['playerName'],"actual":o['actual'],"win":o['win']} for o in outcomes],"loser":"三家","loserId":top['playerId'],"cups":per,"text":text,"revealDice":True}
else:
for o in losers: _drink(find_player(room,o["target"]["playerId"]), base)
d_cups=base*len(winners)
if d_cups: _drink(initiator,d_cups)
loser_id = initiator['id'] if d_cups >= base and d_cups >= (base if losers else 0) else (losers[0]['target']['playerId'] if losers else initiator['id'])
loser_name = initiator['name'] if loser_id==initiator['id'] else next(o['target']['playerName'] for o in losers if o['target']['playerId']==loser_id)
text=f"{initiator['name']} {label}:三家中 {len(winners)} 家赢,{initiator['name']}{d_cups} 杯;{len(losers)} 家输,各喝 {base}"
room["lastResult"]={"multi":True,"targets":[{"name":o['target']['playerName'],"actual":o['actual'],"win":o['win']} for o in outcomes],"loser":loser_name,"loserId":loser_id,"cups":max(d_cups,base if losers else 0),"text":text,"revealDice":True}
wl,dl=multi_outcome(initiator,[o['target']['playerName'] for o in outcomes],drink_map)
room['lastResult']['winners']=wl; room['lastResult']['drinks']=dl
room['lastResult']['calls']=[{"name":o['target']['playerName'],"count":o['target']['count'],"point":o['target']['point'],"mode":o['target'].get('mode',''),"actual":o['actual'],"win":o['win']} for o in outcomes]
room["status"]='result'; room["pendingAction"]=None; append_history(room,'open',room['lastResult']['text'],initiator['name']); room['updatedAt']=now_ms(); return room
def start_multi_direct(code, player_id, kind='open3'):
room=rooms.get((code or "").upper())
if not room: raise ValueError("房间不存在")
if room['status']!='playing': raise ValueError('游戏未开始或已结算')
if room.get('pendingAction'): raise ValueError('当前有待处理的开骰/劈')
initiator=validate_turn(room,player_id)
need=3 if kind.endswith('3') else 2
if need==3 and len(room['players'])<4: raise ValueError('开/劈三家仅限4人局或以上')
if need==2 and len(room['players'])<3: raise ValueError('开/劈两家仅限3人局或以上')
targets=recent_call_targets(room, need)
if len(targets)<need: raise ValueError(f'当前不足{need}家可开')
if kind in ['open3','split3']:
base=1 if kind=='open3' else 2
return finalize_multi_direct(room, initiator, targets, base, '开三家' if kind=='open3' else '劈三家')
# 两家:逐家按单家规则累计,但统一收集应答
base=1 if kind=='open2' else 2
room['pendingAction']={"type":kind,"stage":"wait_target","initiatorId":initiator['id'],"initiatorName":initiator['name'],"baseCups":base,"currentIndex":0,"targets":[{"playerId":t['playerId'],"playerName":t['playerName'],"call":t,"status":"pending"} for t in targets]}
append_history(room,'challenge',f"{initiator['name']} {'开两家' if kind=='open2' else '劈两家'},等待应答",initiator['name'])
room['updatedAt']=now_ms(); return room
def respond_multi(code, player_id, response):
room=rooms.get((code or '').upper())
if not room: raise ValueError('房间不存在')
p=room.get('pendingAction')
if not p or p.get('type') not in ['open2','split2']: raise ValueError('当前没有两家应答')
initiator=find_player(room,p['initiatorId']); base=int(p['baseCups']); idx=int(p['currentIndex']); target_info=p['targets'][idx]; target=find_player(room,target_info['playerId'])
if not initiator or not target: raise ValueError('玩家不存在')
if p['stage']=='wait_target':
if player_id!=target['id']: raise ValueError(f"等待 {target['name']} 应答")
if p['type']=='open2':
if response=='no_counter': target_info['status']='resolve'; target_info['stake']=base
elif response=='counter': target.setdefault('stats',{}).setdefault('counters',0); target['stats']['counters']+=1; p['stage']='wait_initiator'; target_info['counterType']='open'; target_info['declineCups']=base; target_info['acceptStake']=base+1; append_history(room,'counter',f"{target['name']} 反劈,等待 {initiator['name']} 受/不受",target['name']); room['updatedAt']=now_ms(); return room
else: raise ValueError('开两家只支持不反/反劈')
else:
if response=='decline': target.setdefault('stats',{}).setdefault('declines',0); target['stats']['declines']+=1; target_info['status']='decline'; target_info['cups']=base
elif response=='accept': target_info['status']='resolve'; target_info['stake']=base+1
elif response=='counter': target.setdefault('stats',{}).setdefault('counters',0); target['stats']['counters']+=1; p['stage']='wait_initiator'; target_info['counterType']='split'; target_info['declineCups']=base+1; target_info['acceptStake']=base+2; append_history(room,'counter',f"{target['name']} 反劈,等待 {initiator['name']} 受/不受",target['name']); room['updatedAt']=now_ms(); return room
else: raise ValueError('无效应答')
idx += 1
if idx >= len(p['targets']):
# 统一结算
reveal=False; summary=[]; top_loser=None; top_cups=-1; drink_map={}
def _drink(player, cups): add_drink(player, cups); drink_map[player['name']]=drink_map.get(player['name'],0)+cups
for t in p['targets']:
if t['status']=='decline':
player=find_player(room,t['playerId']); _drink(player,int(t['cups'])); summary.append(f"{player['name']} 不受,喝 {t['cups']}")
if t['cups']>top_cups: top_loser, top_cups = player, t['cups']
elif t['status']=='resolve':
call=t['call']; actual=count_actual(room,call); caller=find_player(room,t['playerId']); loser=initiator if actual>=call['count'] else caller; _drink(loser,int(t['stake'])); reveal=True; summary.append(f"{caller['name']}:实际 {actual}{call['point']}{loser['name']}{t['stake']}")
if int(t['stake'])>top_cups: top_loser, top_cups = loser, int(t['stake'])
room['status']='result'; room['pendingAction']=None; room['lastResult']={"multi":True,"targets":[{"name":t['playerName'],"status":t['status']} for t in p['targets']],"loser":top_loser['name'] if top_loser else initiator['name'],"loserId":top_loser['id'] if top_loser else initiator['id'],"cups":top_cups if top_cups>0 else 0,"text":""}
wl,dl=multi_outcome(initiator,[t['playerName'] for t in p['targets']],drink_map); room['lastResult']['winners']=wl; room['lastResult']['drinks']=dl
room['lastResult']['calls']=multi_calls(room, p['targets'])
room['lastResult']['text']=('开两家' if p['type']=='open2' else '劈两家')+''+''.join(summary)
room['lastResult']['revealDice']=reveal
append_history(room,'open',room['lastResult']['text'],initiator['name']); room['updatedAt']=now_ms(); return room
p['currentIndex']=idx; room['updatedAt']=now_ms(); return room
if p['stage']=='wait_initiator':
if player_id!=initiator['id']: raise ValueError(f"等待 {initiator['name']} 受/不受")
if response=='decline': initiator.setdefault('stats',{}).setdefault('declines',0); initiator['stats']['declines']+=1; target_info['status']='decline_initiator'; target_info['cups']=int(target_info['declineCups'])
elif response=='accept': target_info['status']='resolve'; target_info['stake']=int(target_info['acceptStake'])
else: raise ValueError('只能选择受/不受')
p['stage']='wait_target'; p['currentIndex']=idx+1
if p['currentIndex'] >= len(p['targets']):
summary=[]; reveal=False; top_loser=None; top_cups=-1; drink_map={}
def _drink(player, cups): add_drink(player, cups); drink_map[player['name']]=drink_map.get(player['name'],0)+cups
for t in p['targets']:
if t['status'] in ['decline','decline_initiator']:
loser=find_player(room,initiator['id'] if t['status']=='decline_initiator' else t['playerId']); _drink(loser,int(t['cups'])); summary.append(f"{loser['name']} 不受,喝 {t['cups']}")
if int(t['cups'])>top_cups: top_loser, top_cups = loser, int(t['cups'])
elif t['status']=='resolve':
call=t['call']; actual=count_actual(room,call); caller=find_player(room,t['playerId']); loser=initiator if actual>=call['count'] else caller; _drink(loser,int(t['stake'])); reveal=True; summary.append(f"{caller['name']}:实际 {actual}{call['point']}{loser['name']}{t['stake']}")
if int(t['stake'])>top_cups: top_loser, top_cups = loser, int(t['stake'])
room['status']='result'; room['pendingAction']=None; room['lastResult']={"multi":True,"targets":[{"name":t['playerName'],"status":t['status']} for t in p['targets']],"loser":top_loser['name'] if top_loser else initiator['name'],"loserId":top_loser['id'] if top_loser else initiator['id'],"cups":top_cups if top_cups>0 else 0,"text":""}
wl,dl=multi_outcome(initiator,[t['playerName'] for t in p['targets']],drink_map); room['lastResult']['winners']=wl; room['lastResult']['drinks']=dl
room['lastResult']['calls']=multi_calls(room, p['targets'])
room['lastResult']['text']=('开两家' if p['type']=='open2' else '劈两家')+''+''.join(summary)
room['lastResult']['revealDice']=reveal
append_history(room,'open',room['lastResult']['text'],initiator['name']); room['updatedAt']=now_ms(); return room
room['updatedAt']=now_ms(); return room
raise ValueError('无效两家应答')
def bot_turn(code):
room=rooms.get((code or "").upper())
if not room: raise ValueError("房间不存在")
if room["status"]!="playing": return room
pending=room.get("pendingAction")
if pending:
if pending.get("type") in ["open2","split2"]:
idx=int(pending.get("currentIndex",0)); targets=pending.get("targets",[])
if idx < len(targets):
actor_id = targets[idx]["playerId"] if pending["stage"]=="wait_target" else pending["initiatorId"]
actor=find_player(room,actor_id)
if actor and actor.get("bot"):
if pending["stage"]=="wait_target":
if pending['type']=='open2': return respond_multi(code,actor_id,'no_counter' if random.random()<0.72 else 'counter')
return respond_multi(code,actor_id,'accept' if random.random()<0.55 else ('decline' if random.random()<0.78 else 'counter'))
if pending["stage"]=="wait_initiator":
return respond_multi(code,actor_id,'accept' if random.random()<0.62 else 'decline')
return room
actor_id = pending["targetId"] if pending["stage"] in ["wait_caller_counter","wait_caller_accept"] else pending["initiatorId"]
actor=find_player(room,actor_id)
if actor and actor.get("bot"):
if pending["stage"]=="wait_caller_counter": return respond_challenge(code,actor_id,"no_counter" if random.random()<0.75 else "counter")
if pending["stage"]=="wait_caller_accept": return respond_challenge(code,actor_id,"accept" if random.random()<0.55 else ("decline" if random.random()<0.75 else "counter"))
if pending["stage"]=="wait_opener_accept": return respond_challenge(code,actor_id,"accept" if random.random()<0.65 else "decline")
return room
bot=current_player(room)
if not bot or not bot.get("bot"): return room
if room["currentCall"]:
actual=count_actual(room,room["currentCall"]); pressure=room["currentCall"]["count"]-actual
if room["currentCall"]["count"]>=min_count(room)+2:
if pressure>0 and random.random()<0.55: return open_dice(code,bot["id"])
if random.random()<0.16: return open_dice(code,bot["id"])
if room["currentCall"]["count"]>=max_count(room): return open_dice(code,bot["id"])
n=next_call(room,room["currentCall"])
if n["count"]>max_count(room): return open_dice(code,bot["id"])
return call_dice(code,bot["id"],n["count"],n["point"])
def toggle_lock(code,player_id=None):
room=rooms.get((code or "").upper())
if not room: raise ValueError("房间不存在")
if player_id: require_host(room,player_id)
room["locked"]=not room["locked"]; append_history(room,"lock","房间已暂停加入" if room["locked"] else "房间已开放加入"); room["updatedAt"]=now_ms(); return room
def leave_room(code, player_id):
code=(code or "").upper(); room=rooms.get(code)
if not room: raise ValueError("房间不存在")
if room["status"]=="playing": raise ValueError("游戏中不能退出,请等待本局结算")
p=require_player(room,player_id); room["players"]=[x for x in room["players"] if x["id"]!=player_id]
append_history(room,"leave",f"{p['name']} 离开房间",p["name"]); transfer_host_if_needed(room); room["turnIndex"]=0; room["updatedAt"]=now_ms(); deleted=cleanup_room(code); return None if deleted else room
def kick_player(code, host_id, target_id):
code=(code or "").upper(); room=rooms.get(code)
if not room: raise ValueError("房间不存在")
require_host(room,host_id)
if room["status"]=="playing": raise ValueError("游戏中不能踢人")
if host_id==target_id: raise ValueError("不能踢自己,请使用退出房间")
target=find_player(room,target_id)
if not target: raise ValueError("目标玩家不在房间")
room["players"]=[p for p in room["players"] if p["id"]!=target_id]
append_history(room,"kick",f"{target['name']} 被房主移出房间",target["name"]); transfer_host_if_needed(room); room["turnIndex"]=0; room["updatedAt"]=now_ms(); return room
# ---------------- WebSocket纯标准库实现 RFC6455----------------
WS_GUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"
state_lock = threading.RLock() # 保护 rooms / ws_clients 的并发修改
ws_clients = {} # cid -> {"sock","code","player_id","send_lock"}
def recv_exact(sock, n):
buf = b""
while len(buf) < n:
chunk = sock.recv(n - len(buf))
if not chunk: return None
buf += chunk
return buf
def ws_read_frame(sock):
"""读取一帧,返回 (opcode, payload);连接关闭或异常返回 (None, None)。客户端帧带掩码,需解掩。"""
head = recv_exact(sock, 2)
if not head: return None, None
opcode = head[0] & 0x0f
masked = head[1] & 0x80
length = head[1] & 0x7f
if length == 126:
ext = recv_exact(sock, 2)
if not ext: return None, None
length = struct.unpack(">H", ext)[0]
elif length == 127:
ext = recv_exact(sock, 8)
if not ext: return None, None
length = struct.unpack(">Q", ext)[0]
mask = recv_exact(sock, 4) if masked else b"\x00\x00\x00\x00"
if mask is None: return None, None
payload = recv_exact(sock, length) if length else b""
if payload is None: return None, None
if masked:
payload = bytes(payload[i] ^ mask[i % 4] for i in range(len(payload)))
return opcode, payload
def ws_build_frame(payload, opcode=0x1):
data = payload if isinstance(payload, bytes) else payload.encode()
n = len(data)
header = bytes([0x80 | opcode])
if n < 126:
header += bytes([n])
elif n < 65536:
header += bytes([126]) + struct.pack(">H", n)
else:
header += bytes([127]) + struct.pack(">Q", n)
return header + data
def ws_send(client, obj):
frame = ws_build_frame(json.dumps(obj, ensure_ascii=False))
try:
with client["send_lock"]:
client["sock"].sendall(frame)
return True
except Exception:
return False
def ws_handshake(handler):
key = handler.headers.get("Sec-WebSocket-Key")
if not key: return False
accept = base64.b64encode(hashlib.sha1((key + WS_GUID).encode()).digest()).decode()
resp = ("HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\n"
"Connection: Upgrade\r\nSec-WebSocket-Accept: " + accept + "\r\n\r\n")
try:
handler.connection.sendall(resp.encode()); return True
except Exception:
return False
def room_clients(code):
code = (code or "").upper()
with state_lock:
return [c for c in ws_clients.values() if c.get("code") == code]
def lobby_clients():
with state_lock:
return [c for c in ws_clients.values() if not c.get("code")]
def broadcast_room(code):
"""向房内所有连接推送各自视角的房间状态;房间已删除则推 closed。"""
code = (code or "").upper(); room = rooms.get(code)
for c in room_clients(code):
if room: ws_send(c, {"type": "room", "room": public_room(room, c.get("player_id"))})
else: ws_send(c, {"type": "closed"})
def broadcast_rooms():
payload = {"type": "rooms", "rooms": [public_room(r) for r in rooms.values()]}
for c in lobby_clients():
ws_send(c, payload)
def schedule_bots(code):
threading.Timer(0.9, lambda: run_bots(code)).start()
def run_bots(code):
"""服务端驱动机器人:推进一步,若有变化则广播并继续排程,直到轮到真人或本局结束。"""
code = (code or "").upper(); changed = False
with state_lock:
room = rooms.get(code)
if room and room["status"] == "playing":
before = room["updatedAt"]
try: bot_turn(code)
except Exception as e: print("bot error:", e)
changed = rooms.get(code, {}).get("updatedAt") != before
if changed:
broadcast_room(code); schedule_bots(code)
# 推进游戏进程、需触发机器人接续的动作
BOT_TRIGGER = {"start","call","open","split","steal","multi","respond","respondMulti"}
def handle_ws(client, msg):
t = msg.get("type")
try:
acted = None; room = None
with state_lock:
if t == "rooms":
ws_send(client, {"type": "rooms", "rooms": [public_room(r) for r in rooms.values()]}); return
elif t == "create":
room, player = create_room(msg.get("game", "dice"), msg.get("name", "玩家"), msg.get("playerId"))
client["code"] = room["code"]; client["player_id"] = player["id"]; acted = room["code"]
ws_send(client, {"type": "joined", "room": public_room(room, player["id"]), "player": player})
elif t == "join":
room, player, reused = join_room(msg.get("code", ""), msg.get("name", "玩家"), msg.get("playerId"))
client["code"] = room["code"]; client["player_id"] = player["id"]; acted = room["code"]
ws_send(client, {"type": "joined", "room": public_room(room, player["id"]), "player": player, "reused": reused})
else:
code = client.get("code"); pid = client.get("player_id")
if not code: ws_send(client, {"type": "error", "error": "尚未加入房间"}); return
acted = code
if t == "ready": room = set_ready(code, pid, msg.get("ready"))
elif t == "start": room = start_room(code, pid)
elif t == "call": room = call_dice(code, pid, msg.get("count", 1), msg.get("point", 1), msg.get("mode"))
elif t == "open": room = open_dice(code, pid)
elif t == "split": room = split_dice(code, pid)
elif t == "steal": room = steal_split(code, pid)
elif t == "multi": room = start_multi_direct(code, pid, msg.get("kind", "open2"))
elif t == "respond": room = respond_challenge(code, pid, msg.get("response", ""))
elif t == "respondMulti": room = respond_multi(code, pid, msg.get("response", ""))
elif t == "settings": room = update_settings(code, pid, msg.get("cupsPerLoss"))
elif t == "lock": room = toggle_lock(code, pid)
elif t == "kick": room = kick_player(code, pid, msg.get("targetId"))
elif t == "addBot": room = add_bot(code, pid)
elif t == "leave":
room = leave_room(code, pid); client["code"] = None; client["player_id"] = None
ws_send(client, {"type": "left"})
else:
ws_send(client, {"type": "error", "error": "未知消息"}); return
if acted:
broadcast_room(acted); broadcast_rooms()
if t in BOT_TRIGGER: schedule_bots(acted)
except Exception as e:
ws_send(client, {"type": "error", "error": str(e)})
def ws_serve(handler):
sock = handler.connection
cid = new_id("w")
client = {"sock": sock, "code": None, "player_id": None, "send_lock": threading.Lock()}
with state_lock:
ws_clients[cid] = client
ws_send(client, {"type": "rooms", "rooms": [public_room(r) for r in rooms.values()]})
try:
while True:
opcode, payload = ws_read_frame(sock)
if opcode is None or opcode == 0x8: break # 关闭
if opcode == 0x9: # ping -> pong
try:
with client["send_lock"]: sock.sendall(ws_build_frame(payload, 0xA))
except Exception: break
continue
if opcode == 0xA: continue # pong
if opcode != 0x1: continue # 仅处理文本帧
try: data = json.loads(payload.decode() or "{}")
except Exception: continue
handle_ws(client, data)
finally:
with state_lock:
ws_clients.pop(cid, None)
class Handler(SimpleHTTPRequestHandler):
def __init__(self,*a,**kw): super().__init__(*a,directory=str(ROOT),**kw)
def log_message(self,fmt,*a): print("[%s] %s"%(self.log_date_time_string(),fmt%a))
def end_headers(self):
# 禁用静态文件缓存,避免浏览器用旧版 live.html/JS/CSSWS 握手走 sendall不经此处
self.send_header("Cache-Control","no-cache, no-store, must-revalidate")
super().end_headers()
def do_OPTIONS(self): self.send_response(204); self.send_header("Access-Control-Allow-Origin","*"); self.send_header("Access-Control-Allow-Methods","GET, POST, OPTIONS"); self.send_header("Access-Control-Allow-Headers","Content-Type"); self.end_headers()
def do_GET(self):
p=urlparse(self.path); qs=parse_qs(p.query)
if p.path=="/ws" and self.headers.get("Upgrade","").lower()=="websocket":
self.close_connection=True
if ws_handshake(self): ws_serve(self)
return
if p.path=="/api/rooms": send_json(self,{"ok":True,"rooms":[public_room(r) for r in rooms.values()]}); return
if p.path=="/api/room":
code=qs.get("code",[""])[0].upper(); viewer=qs.get("playerId",[None])[0]; room=rooms.get(code)
send_json(self,{"ok":bool(room),"room":public_room(room,viewer) if room else None,"error":None if room else "房间不存在"},200 if room else 404); return
return super().do_GET()
def do_POST(self):
try:
p=urlparse(self.path); d=read_json(self); viewer=d.get("playerId")
if p.path=="/api/create":
room,player=create_room(d.get("game","dice"),d.get("name","玩家"),viewer); send_json(self,{"ok":True,"room":public_room(room,player["id"]),"player":player}); return
if p.path=="/api/join":
room,player,reused=join_room(d.get("code",""),d.get("name","玩家"),viewer); send_json(self,{"ok":True,"room":public_room(room,player["id"]),"player":player,"reused":reused}); return
if p.path=="/api/ready": send_json(self,{"ok":True,"room":public_room(set_ready(d.get("code",""),viewer,d.get("ready",None)),viewer)}); return
if p.path=="/api/add-bot": send_json(self,{"ok":True,"room":public_room(add_bot(d.get("code",""),viewer),viewer)}); return
if p.path=="/api/settings": send_json(self,{"ok":True,"room":public_room(update_settings(d.get("code",""),viewer,d.get("cupsPerLoss",None)),viewer)}); return
if p.path=="/api/leave":
room=leave_room(d.get("code",""),viewer); send_json(self,{"ok":True,"room":public_room(room,viewer) if room else None,"deleted":room is None}); return
if p.path=="/api/kick": send_json(self,{"ok":True,"room":public_room(kick_player(d.get("code",""),viewer,d.get("targetId")),viewer)}); return
if p.path=="/api/start": send_json(self,{"ok":True,"room":public_room(start_room(d.get("code",""),viewer),viewer)}); return
if p.path=="/api/call": send_json(self,{"ok":True,"room":public_room(call_dice(d.get("code",""),viewer,d.get("count",1),d.get("point",1),d.get("mode",None)),viewer)}); return
if p.path=="/api/open": send_json(self,{"ok":True,"room":public_room(open_dice(d.get("code",""),viewer),viewer)}); return
if p.path=="/api/split": send_json(self,{"ok":True,"room":public_room(split_dice(d.get("code",""),viewer),viewer)}); return
if p.path=="/api/steal-split": send_json(self,{"ok":True,"room":public_room(steal_split(d.get("code",""),viewer),viewer)}); return
if p.path=="/api/multi-open2": send_json(self,{"ok":True,"room":public_room(start_multi_direct(d.get("code",""),viewer,'open2'),viewer)}); return
if p.path=="/api/multi-split2": send_json(self,{"ok":True,"room":public_room(start_multi_direct(d.get("code",""),viewer,'split2'),viewer)}); return
if p.path=="/api/multi-open3": send_json(self,{"ok":True,"room":public_room(start_multi_direct(d.get("code",""),viewer,'open3'),viewer)}); return
if p.path=="/api/multi-split3": send_json(self,{"ok":True,"room":public_room(start_multi_direct(d.get("code",""),viewer,'split3'),viewer)}); return
if p.path=="/api/respond-challenge": send_json(self,{"ok":True,"room":public_room(respond_challenge(d.get("code",""),viewer,d.get("response","")),viewer)}); return
if p.path=="/api/respond-multi": send_json(self,{"ok":True,"room":public_room(respond_multi(d.get("code",""),viewer,d.get("response","")),viewer)}); return
if p.path=="/api/bot-turn": send_json(self,{"ok":True,"room":public_room(bot_turn(d.get("code","")),viewer)}); return
if p.path=="/api/toggle-lock": send_json(self,{"ok":True,"room":public_room(toggle_lock(d.get("code",""),viewer),viewer)}); return
send_json(self,{"ok":False,"error":"未知接口"},404)
except Exception as e: send_json(self,{"ok":False,"error":str(e)},400)
if __name__=="__main__":
create_room("dice","演示房主")
port=8765
print(f"酒桌派对开发服务器http://127.0.0.1:{port}/live.html WebSocket: /ws")
server=ThreadingHTTPServer(("0.0.0.0",port),Handler)
server.daemon_threads=True
server.serve_forever()

1
style.css Normal file

File diff suppressed because one or more lines are too long

43
test_action_gating.py Normal file
View File

@@ -0,0 +1,43 @@
import json, urllib.request
base='http://127.0.0.1:8765'
def post(path, data):
req=urllib.request.Request(base+path, data=json.dumps(data).encode(), headers={'Content-Type':'application/json'})
try:
return json.loads(urllib.request.urlopen(req, timeout=5).read())
except Exception as e:
body=e.read().decode() if hasattr(e,'read') else str(e)
return json.loads(body) if body.startswith('{') else {'ok':False,'error':body}
def get_room(code):
return json.loads(urllib.request.urlopen(base+'/api/room?code='+code).read())['room']
# 复刻前端 renderActions 的近似history 里不同叫骰人数
def recent_caller_count(room):
s=set()
for h in room.get('history',[]):
if h.get('type')=='call' and h.get('player'): s.add(h['player'])
return len(s)
# 4 人局:房主 + 3 机器人
r=post('/api/create', {'game':'dice','name':'小鱼','playerId':'p-host'})
code=r['room']['code']
for _ in range(3): post('/api/add-bot', {'code':code,'playerId':'p-host'})
post('/api/start', {'code':code})
# 房主先叫,再让 3 个机器人各叫一手,回到房主回合
post('/api/call', {'code':code,'playerId':'p-host','count':4,'point':1})
for _ in range(3): post('/api/bot-turn', {'code':code,'playerId':'p-host'})
room=get_room(code)
callers=recent_caller_count(room)
my_turn = room['turnPlayerId']=='p-host'
frontend_shows_open3 = my_turn and len(room['players'])>=4 and callers>=3
print('turn=', room['turnPlayerName'], 'distinct callers=', callers, 'frontend_shows_open3=', frontend_shows_open3)
assert frontend_shows_open3, '前端门控未点亮开三家,测试前提不成立'
# 前端点亮了 → 服务端必须接受
res=post('/api/multi-open3', {'code':code,'playerId':'p-host'})
print('server multi-open3 ok=', res['ok'], '| status=', res.get('room',{}).get('status') if res['ok'] else res.get('error'))
assert res['ok'], '前端亮了开三家但服务端拒绝——门控与服务端不一致'
assert res['room']['status']=='result', '开三家应直接结算'
print('result:', res['room']['lastResult']['text'])
print('PASS: 动作门控与服务端接受判定一致(开三家)')

53
test_bot_multi.py Normal file
View File

@@ -0,0 +1,53 @@
import json, urllib.request
base='http://127.0.0.1:8765'
def post(path, data):
req=urllib.request.Request(base+path, data=json.dumps(data).encode(), headers={'Content-Type':'application/json'})
try:
return json.loads(urllib.request.urlopen(req, timeout=5).read())
except Exception as e:
body=e.read().decode() if hasattr(e,'read') else str(e)
return json.loads(body) if body.startswith('{') else {'ok':False,'error':body}
def get_room(code):
return json.loads(urllib.request.urlopen(base+'/api/room?code='+code).read())['room']
def actor_id(p):
if p['type'] in ('open2','split2'):
idx=p.get('currentIndex',0); targets=p.get('targets',[])
return targets[idx]['playerId'] if (p['stage']=='wait_target' and idx<len(targets)) else p['initiatorId']
return p['targetId'] if p['stage'] in ('wait_caller_counter','wait_caller_accept') else p['initiatorId']
# 房主(真人) + 2 机器人
r=post('/api/create', {'game':'dice','name':'小鱼','playerId':'p-host'})
code=r['room']['code']
post('/api/add-bot', {'code':code,'playerId':'p-host'})
post('/api/add-bot', {'code':code,'playerId':'p-host'})
post('/api/start', {'code':code})
# 房主先叫 3 个 1再让两个机器人各叫一手
post('/api/call', {'code':code,'playerId':'p-host','count':3,'point':1})
post('/api/bot-turn', {'code':code,'playerId':'p-host'}) # bot1 叫
post('/api/bot-turn', {'code':code,'playerId':'p-host'}) # bot2 叫
rm=get_room(code)
print('before open2: turn=', rm['turnPlayerName'], 'calls=', len([1 for h in rm['history'] if h['type']=='call']))
# 房主发起开两家
op=post('/api/multi-open2', {'code':code,'playerId':'p-host'})
assert op['ok'], '发起开两家失败: '+str(op.get('error'))
print('open2 start: type=', op['room']['pendingAction']['type'])
# 模拟前端 maybeBotTurn应战者是机器人就触发 bot-turn轮到真人则代答
room=op['room']
for _ in range(10):
p=room.get('pendingAction')
if not p: break
actor=next((x for x in room['players'] if x['id']==actor_id(p)), None)
if actor and actor.get('bot'):
room=post('/api/bot-turn', {'code':code,'playerId':'p-host'})['room']
else:
resp='accept' if p['stage']=='wait_initiator' else 'no_counter'
room=post('/api/respond-multi', {'code':code,'playerId':'p-host','response':resp})['room']
print('after responses: status=', room['status'])
print('result:', room.get('lastResult',{}).get('text'))
assert room['status']=='result', '开两家未结算——机器人未自动应战'
print('PASS: 机器人在开两家中自动应战并完成结算')

35
test_challenge.py Normal file
View File

@@ -0,0 +1,35 @@
import importlib.util
spec=importlib.util.spec_from_file_location('srv','/var/minis/workspace/drinking-games-ui/server.py')
srv=importlib.util.module_from_spec(spec); spec.loader.exec_module(srv)
def setup():
room, host=srv.create_room('dice','房主','p-host')
room,p2,_=srv.join_room(room['code'],'客人','p-guest')
srv.set_ready(room['code'],'p-guest',True)
srv.start_room(room['code'],'p-host')
srv.call_dice(room['code'],'p-host',2,1,'')
return room
# 普通开骰 -> 反劈 -> 不受
room=setup(); code=room['code']
srv.open_dice(code,'p-guest')
print('open pending', room['pendingAction']['stage'])
srv.respond_challenge(code,'p-host','counter')
print('counter pending', room['pendingAction']['stage'])
srv.respond_challenge(code,'p-guest','decline')
print('open counter decline result', room['status'], room['lastResult']['loser'], room['lastResult']['cups'], room['lastResult']['revealDice'])
# 劈 -> 不受
room=setup(); code=room['code']
srv.split_dice(code,'p-guest')
print('split pending', room['pendingAction']['stage'])
srv.respond_challenge(code,'p-host','decline')
print('split decline result', room['status'], room['lastResult']['loser'], room['lastResult']['cups'], room['lastResult']['revealDice'])
# 劈 -> 反劈 -> 受
room=setup(); code=room['code']
srv.split_dice(code,'p-guest')
srv.respond_challenge(code,'p-host','counter')
print('split counter pending', room['pendingAction']['stage'])
srv.respond_challenge(code,'p-guest','accept')
print('split counter accept result', room['status'], room['lastResult']['cups'], room['lastResult']['revealDice'])

30
test_dice_enhanced.py Normal file
View File

@@ -0,0 +1,30 @@
import json, urllib.request, time
base='http://127.0.0.1:8765'
def post(path, data):
req=urllib.request.Request(base+path, data=json.dumps(data).encode(), headers={'Content-Type':'application/json'})
try: return json.loads(urllib.request.urlopen(req, timeout=5).read())
except Exception as e:
body=e.read().decode() if hasattr(e,'read') else str(e)
try: return json.loads(body)
except Exception: return {'ok':False,'error':body}
time.sleep(1)
r=post('/api/create', {'game':'dice','name':'房主','playerId':'p-host'})
code=r['room']['code']
print('create', r['ok'], code, r['player']['ready'])
j=post('/api/join', {'code':code,'name':'客人','playerId':'p-guest'})
print('join guest ready', j['room']['players'][1]['ready'])
bad=post('/api/start', {'code':code,'playerId':'p-host'})
print('start before ready', bad['ok'], bad.get('error'))
ready=post('/api/ready', {'code':code,'playerId':'p-guest'})
print('guest ready', [p['ready'] for p in ready['room']['players']])
setg=post('/api/settings', {'code':code,'playerId':'p-host','cupsPerLoss':2})
print('settings', setg['room']['settings'])
bot=post('/api/add-bot', {'code':code,'playerId':'p-host'})
print('add bot', len(bot['room']['players']))
s=post('/api/start', {'code':code,'playerId':'p-host'})
print('start', s['ok'], s['room']['status'], s['room']['diceRule'])
# 3人局最低3个1
badcall=post('/api/call', {'code':code,'playerId':'p-host','count':2,'point':6})
print('bad call', badcall['ok'], badcall.get('error'))
c=post('/api/call', {'code':code,'playerId':'p-host','count':3,'point':1})
print('call', c['ok'], c['room']['currentCall'])

33
test_explicit_mode.py Normal file
View File

@@ -0,0 +1,33 @@
import importlib.util
spec=importlib.util.spec_from_file_location('srv','/var/minis/workspace/drinking-games-ui/server.py')
srv=importlib.util.module_from_spec(spec); spec.loader.exec_module(srv)
def new_room():
room, host=srv.create_room('dice','房主','p-host')
for _ in range(2): srv.add_bot(room['code'],'p-host')
srv.start_room(room['code'],'p-host')
return room
room=new_room()
srv.call_dice(room['code'],'p-host',5,2,'')
print('first 5x2 zhai', room['currentCall'])
assert room['currentCall']['mode']==''
room=new_room()
srv.call_dice(room['code'],'p-host',5,2,'')
print('first 5x2 fly', room['currentCall'])
assert room['currentCall']['mode']==''
room=new_room()
try:
srv.call_dice(room['code'],'p-host',5,1,'')
print('bad 5x1 fly accepted')
except Exception as e:
print('bad 5x1 fly rejected', str(e))
room=new_room()
try:
srv.call_dice(room['code'],'p-host',3,2,'')
print('bad 3x2 accepted')
except Exception as e:
print('bad 3x2 rejected', str(e))

15
test_first_call.py Normal file
View File

@@ -0,0 +1,15 @@
import importlib.util
spec=importlib.util.spec_from_file_location('srv','/var/minis/workspace/drinking-games-ui/server.py')
srv=importlib.util.module_from_spec(spec); spec.loader.exec_module(srv)
room, host=srv.create_room('dice','房主','p-host')
for _ in range(2): srv.add_bot(room['code'],'p-host')
srv.start_room(room['code'],'p-host')
print('players', len(room['players']), 'min', srv.min_count(room))
try:
srv.call_dice(room['code'],'p-host',3,2)
print('bad accepted')
except Exception as e:
print('bad rejected', str(e))
room['turnIndex']=0
srv.call_dice(room['code'],'p-host',3,1)
print('ok', room['currentCall'])

23
test_flow.py Normal file
View File

@@ -0,0 +1,23 @@
import json, urllib.request, time
base='http://127.0.0.1:8765'
def post(path, data):
req=urllib.request.Request(base+path, data=json.dumps(data).encode(), headers={'Content-Type':'application/json'})
try:
return json.loads(urllib.request.urlopen(req, timeout=5).read())
except Exception as e:
body=e.read().decode() if hasattr(e,'read') else str(e)
return json.loads(body) if body.startswith('{') else {'ok':False,'error':body}
time.sleep(1)
r=post('/api/create', {'game':'dice','name':'小鱼','playerId':'p-test'})
print('create', r['ok'], r['room']['code'], len(r['room']['players']))
code=r['room']['code']
j=post('/api/join', {'code':code,'name':'小鱼','playerId':'p-test'})
print('rejoin', j['ok'], j.get('reused'), len(j['room']['players']))
s=post('/api/start', {'code':code})
print('start', s['ok'], s['room']['status'], s['room']['turnPlayerName'], len(s['room']['players']))
c1=post('/api/call', {'code':code,'playerId':'p-test','count':1,'point':1})
print('call1', c1['ok'], c1['room']['currentCall'], 'turn', c1['room']['turnPlayerName'])
bad=post('/api/call', {'code':code,'playerId':'p-test','count':1,'point':1})
print('bad call', bad['ok'], bad.get('error'))
op=post('/api/open', {'code':code,'playerId':'p-test'})
print('open', op['ok'], op['room']['status'], op['room']['lastResult']['loser'])

37
test_multi_play.py Normal file
View File

@@ -0,0 +1,37 @@
import importlib.util
spec=importlib.util.spec_from_file_location('srv','/var/minis/workspace/drinking-games-ui/server.py')
srv=importlib.util.module_from_spec(spec)
spec.loader.exec_module(srv)
room, host=srv.create_room('dice','房主','p-host')
room,p2,_=srv.join_room(room['code'],'A','p-a')
room,p3,_=srv.join_room(room['code'],'B','p-b')
room,p4,_=srv.join_room(room['code'],'C','p-c')
for pid in ['p-a','p-b','p-c']:
srv.set_ready(room['code'],pid,True)
srv.start_room(room['code'],'p-host')
room['turnIndex']=1; srv.call_dice(room['code'],'p-a',4,1,'')
room['turnIndex']=2; srv.call_dice(room['code'],'p-b',5,2,'')
room['turnIndex']=3; srv.call_dice(room['code'],'p-c',5,3,'')
room['turnIndex']=0
print('recent3', [t['playerName'] for t in srv.recent_call_targets(room,3)])
r=srv.start_multi_direct(room['code'],'p-host','open3')
print('open3', r['status'], r['lastResult']['multi'], r['lastResult']['revealDice'], r['lastResult']['text'])
room, host=srv.create_room('dice','房主','h')
room,p2,_=srv.join_room(room['code'],'A','a')
room,p3,_=srv.join_room(room['code'],'B','b')
for pid in ['a','b']:
srv.set_ready(room['code'],pid,True)
srv.start_room(room['code'],'h')
room['turnIndex']=1; srv.call_dice(room['code'],'a',3,1,'')
room['turnIndex']=2; srv.call_dice(room['code'],'b',4,2,'')
room['turnIndex']=0
srv.start_multi_direct(room['code'],'h','open2')
print('open2 pending', room['pendingAction']['type'], [t['playerName'] for t in room['pendingAction']['targets']])
srv.respond_multi(room['code'],'b','no_counter')
print('after first target idx', room['pendingAction']['currentIndex'])
srv.respond_multi(room['code'],'a','counter')
print('after counter stage', room['pendingAction']['stage'])
srv.respond_multi(room['code'],'h','decline')
print('open2 done', room['status'], room['lastResult']['text'])

52
test_multi_result.py Normal file
View File

@@ -0,0 +1,52 @@
import json, urllib.request
base='http://127.0.0.1:8765'
def post(path, data):
req=urllib.request.Request(base+path, data=json.dumps(data).encode(), headers={'Content-Type':'application/json'})
try: return json.loads(urllib.request.urlopen(req, timeout=5).read())
except Exception as e:
body=e.read().decode() if hasattr(e,'read') else str(e)
return json.loads(body) if body.startswith('{') else {'ok':False,'error':body}
def get_room(code): return json.loads(urllib.request.urlopen(base+'/api/room?code='+code).read())['room']
# 4 人局:房主 + 3 机器人,房主开三家直接结算
r=post('/api/create', {'game':'dice','name':'小鱼','playerId':'p-host'})
code=r['room']['code']
for _ in range(3): post('/api/add-bot', {'code':code,'playerId':'p-host'})
post('/api/start', {'code':code})
post('/api/call', {'code':code,'playerId':'p-host','count':4,'point':1})
for _ in range(3): post('/api/bot-turn', {'code':code,'playerId':'p-host'})
res=post('/api/multi-open3', {'code':code,'playerId':'p-host'})
assert res['ok'], '开三家失败: '+str(res.get('error'))
lr=res['room']['lastResult']
players=[p['name'] for p in res['room']['players']]
print('text:', lr['text'])
print('winners:', lr.get('winners'))
print('drinks :', lr.get('drinks'))
print('calls :', lr.get('calls'))
# 1) 结构完整winners/drinks/calls 都存在
assert 'winners' in lr and 'drinks' in lr, '缺少 winners/drinks 字段'
# 0) 叫点明细:每个被开的参与者都带 count/point/mode 与实际数
calls=lr.get('calls') or []
assert calls, '缺少 calls 叫点明细'
assert len(calls)==3, '开三家应有 3 家叫点明细'
for c in calls:
assert c.get('count') and c.get('point') and c.get('mode'), f"叫点明细缺字段: {c}"
assert c.get('actual') is not None, f"开三家应有实际数: {c}"
assert isinstance(c.get('win'), bool), f"开三家应有成立判定: {c}"
assert {c['name'] for c in calls}==set(players)-{'小鱼'}, '开三家叫点明细应为发起者外的 3 名叫骰人'
drink_names=[d['name'] for d in lr['drinks']]
# 2) 每个喝酒的人杯数 > 0
assert all(d['cups']>0 for d in lr['drinks']), '存在 0 杯的喝酒项'
# 3) 赢家与喝酒者不重叠,且二者并集覆盖全部 4 名参与者(不再只显示一个人)
assert set(lr['winners']) & set(drink_names)==set(), '赢家与喝酒者重叠'
assert set(lr['winners']) | set(drink_names)==set(players), '赢家+喝酒者未覆盖全部参与者'
# 4) 喝酒明细与各玩家 cups 增量一致
cups_by_name={p['name']:p['cups'] for p in res['room']['players']}
for d in lr['drinks']:
assert cups_by_name[d['name']]>=d['cups'], f"{d['name']} 杯数不一致"
print(f"参与者 {len(players)} 人 -> 赢家 {len(lr['winners'])} · 喝酒 {len(drink_names)}")
print('PASS: 多家结算输出完整的赢家/各人喝杯明细,覆盖全部参与者')

30
test_multiplayer.py Normal file
View File

@@ -0,0 +1,30 @@
import json, urllib.request, time
base='http://127.0.0.1:8765'
def post(path, data):
req=urllib.request.Request(base+path, data=json.dumps(data).encode(), headers={'Content-Type':'application/json'})
try: return json.loads(urllib.request.urlopen(req, timeout=5).read())
except Exception as e:
body=e.read().decode() if hasattr(e,'read') else str(e)
try: return json.loads(body)
except Exception: return {'ok':False,'error':body}
def get(path): return json.loads(urllib.request.urlopen(base+path, timeout=5).read())
time.sleep(1)
r=post('/api/create', {'game':'dice','name':'房主','playerId':'p-host'})
code=r['room']['code']
print('create', r['ok'], code, r['player']['host'])
j=post('/api/join', {'code':code,'name':'客人','playerId':'p-guest'})
print('join', j['ok'], len(j['room']['players']))
re=post('/api/join', {'code':code,'name':'客人','playerId':'p-guest'})
print('rejoin', re['ok'], re['reused'], len(re['room']['players']))
bad=post('/api/start', {'code':code,'playerId':'p-guest'})
print('guest start', bad['ok'], bad.get('error'))
s=post('/api/start', {'code':code,'playerId':'p-host'})
print('host start', s['ok'], s['room']['status'])
room_host=get('/api/room?code='+code+'&playerId=p-host')['room']
room_guest=get('/api/room?code='+code+'&playerId=p-guest')['room']
print('privacy host dice lens', [len(p['dice']) for p in room_host['players']])
print('privacy guest dice lens', [len(p['dice']) for p in room_guest['players']])
c=post('/api/call', {'code':code,'playerId':'p-host','count':1,'point':1})
print('bad min', c['ok'], c.get('error'))
c=post('/api/call', {'code':code,'playerId':'p-host','count':len(s['room']['players']),'point':1})
print('call min', c['ok'], c['room']['currentCall'], len(c['room']['history']))

23
test_one_zhai.py Normal file
View File

@@ -0,0 +1,23 @@
import importlib.util
spec=importlib.util.spec_from_file_location('srv','/var/minis/workspace/drinking-games-ui/server.py')
srv=importlib.util.module_from_spec(spec); spec.loader.exec_module(srv)
room, host=srv.create_room('dice','房主','p-host')
for _ in range(2): srv.add_bot(room['code'],'p-host')
srv.start_room(room['code'],'p-host')
# 上一手飞叫1必须变斋
room['currentCall']={'count':6,'point':6,'mode':'','by':'Mia','byId':'bot'}
room['turnIndex']=0
srv.call_dice(room['code'],'p-host',7,1,'')
print('7x1 zhai', room['currentCall'])
assert room['currentCall']['mode']==''
# 平斋/切斋后重新按斋规则7个1斋后破斋要14个
room['turnIndex']=0
try:
srv.call_dice(room['code'],'p-host',13,2,'')
print('bad break accepted')
except Exception as e:
print('bad break rejected', str(e))
room['turnIndex']=0
srv.call_dice(room['code'],'p-host',14,2,'')
print('break after ping zhai ok', room['currentCall'])
assert room['currentCall']['mode']==''

30
test_ping_zhai.py Normal file
View File

@@ -0,0 +1,30 @@
import importlib.util
spec=importlib.util.spec_from_file_location('srv','/var/minis/workspace/drinking-games-ui/server.py')
srv=importlib.util.module_from_spec(spec); spec.loader.exec_module(srv)
room, host=srv.create_room('dice','房主','p-host')
for _ in range(2): srv.add_bot(room['code'],'p-host')
srv.start_room(room['code'],'p-host')
# 飞局8个2飞
room['currentCall']={'count':8,'point':2,'mode':'','by':'Mia','byId':'bot'}
room['turnIndex']=0
srv.call_dice(room['code'],'p-host',8,1,'')
print('ping zhai', room['currentCall'])
assert room['currentCall']['mode']==''
room['currentCall']={'count':8,'point':2,'mode':'','by':'Mia','byId':'bot'}
room['turnIndex']=0
srv.call_dice(room['code'],'p-host',8,3,'')
print('continue fly', room['currentCall'])
assert room['currentCall']['mode']==''
room['currentCall']={'count':8,'point':2,'mode':'','by':'Mia','byId':'bot'}
room['turnIndex']=0
try:
srv.call_dice(room['code'],'p-host',7,1,'')
print('bad accepted')
except Exception as e:
print('bad lower rejected', str(e))
# 输家先叫:让房主输,下一局应房主起手
room['status']='playing'; room['turnIndex']=1; room['currentCall']={'count':20,'point':6,'mode':'','by':'房主','byId':'p-host'}
srv.open_dice(room['code'], room['players'][1]['id'])
print('loser', room['lastResult']['loser'])
srv.start_room(room['code'],'p-host')
print('next starter', srv.current_player(room)['name'])

25
test_rule_flow.py Normal file
View File

@@ -0,0 +1,25 @@
import json, urllib.request, time
base='http://127.0.0.1:8765'
def post(path, data):
req=urllib.request.Request(base+path, data=json.dumps(data).encode(), headers={'Content-Type':'application/json'})
try: return json.loads(urllib.request.urlopen(req, timeout=5).read())
except Exception as e:
body=e.read().decode() if hasattr(e,'read') else str(e)
return json.loads(body) if body.startswith('{') else {'ok':False,'error':body}
time.sleep(1)
r=post('/api/create', {'game':'dice','name':'小鱼','playerId':'p-test-rule'})
code=r['room']['code']
s=post('/api/start', {'code':code})
print('players/min/zhai', len(s['room']['players']), s['room']['minCount'], s['room']['zhaiMax'])
print('bad low', post('/api/call', {'code':code,'playerId':'p-test-rule','count':1,'point':1}))
c=post('/api/call', {'code':code,'playerId':'p-test-rule','count':4,'point':1})
print('first', c['ok'], c['room']['currentCall'])
# Let bots rotate until user turn or result
room=c['room']
for _ in range(5):
if room['status']!='playing' or not room['turnIsBot']: break
room=post('/api/bot-turn', {'code':code})['room']
print('after bots', room['status'], room['currentCall'], room['turnPlayerName'], room.get('lastResult'))
if room['status']=='playing':
op=post('/api/open', {'code':code,'playerId':'p-test-rule'})
print('open', op['ok'], op['room']['lastResult']['text'], len(op['room']['lastResult']['allDice']))

25
test_rules.py Normal file
View File

@@ -0,0 +1,25 @@
import json, urllib.request, time
base='http://127.0.0.1:8765'
def post(path, data):
req=urllib.request.Request(base+path, data=json.dumps(data).encode(), headers={'Content-Type':'application/json'})
try:
return json.loads(urllib.request.urlopen(req, timeout=5).read())
except Exception as e:
body=e.read().decode() if hasattr(e,'read') else str(e)
try: return json.loads(body)
except Exception: return {'ok':False,'error':body}
time.sleep(1)
r=post('/api/create', {'game':'dice','name':'小鱼','playerId':'p-test'})
code=r['room']['code']
s=post('/api/start', {'code':code})
print('players', len(s['room']['players']), 'min', s['room']['diceRule']['minCount'], 'flyFrom', s['room']['diceRule']['flyFrom'])
print('bad-low', post('/api/call', {'code':code,'playerId':'p-test','count':3,'point':6}).get('error'))
c1=post('/api/call', {'code':code,'playerId':'p-test','count':4,'point':1})
print('call1', c1['ok'], c1['room']['currentCall'])
# 如果机器人连续轮转后又轮到真人,当前叫点应被机器人加到更高
for i in range(5):
room=post('/api/bot-turn', {'code':code})['room']
print('turn', i, room['turnPlayerName'], room['currentCall'], room['status'])
if room['status']=='result' or not room['turnPlayerBot']:
break
print('final', room['status'], room['turnPlayerName'])

109
test_ws.py Normal file
View File

@@ -0,0 +1,109 @@
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 端到端通过(连接/创建/加入/广播/服务端机器人驱动)')

17
test_zhai_continue.py Normal file
View File

@@ -0,0 +1,17 @@
import importlib.util
spec=importlib.util.spec_from_file_location('srv','/var/minis/workspace/drinking-games-ui/server.py')
srv=importlib.util.module_from_spec(spec); spec.loader.exec_module(srv)
room, host=srv.create_room('dice','房主','p-host')
for _ in range(2): srv.add_bot(room['code'],'p-host')
srv.start_room(room['code'],'p-host')
# 模拟上一手4个4斋
room['currentCall']={'count':4,'point':4,'mode':'','by':'Mia','byId':'bot'}
room['turnIndex']=0
srv.call_dice(room['code'],'p-host',5,2)
print('5x2 after 4x4', room['currentCall'])
assert room['currentCall']['mode']==''
room['currentCall']={'count':4,'point':4,'mode':'','by':'Mia','byId':'bot'}
room['turnIndex']=0
srv.call_dice(room['code'],'p-host',8,2)
print('8x2 after 4x4', room['currentCall'])
assert room['currentCall']['mode']==''

24
test_zhai_rule.py Normal file
View File

@@ -0,0 +1,24 @@
import importlib.util
spec=importlib.util.spec_from_file_location('srv','/var/minis/workspace/drinking-games-ui/server.py')
srv=importlib.util.module_from_spec(spec); spec.loader.exec_module(srv)
room, host=srv.create_room('dice','房主','p-host')
for _ in range(3): srv.add_bot(room['code'],'p-host')
srv.start_room(room['code'],'p-host')
print('players', len(room['players']), srv.min_count(room), srv.next_call(room, None))
srv.call_dice(room['code'],'p-host',4,1)
print('call', room['currentCall'])
room['turnIndex']=0
try:
srv.call_dice(room['code'],'p-host',4,4)
print('bad same count accepted')
except Exception as e:
print('bad same count rejected', str(e))
room['turnIndex']=0
try:
srv.call_dice(room['code'],'p-host',6,2)
print('bad break accepted')
except Exception as e:
print('bad break rejected', str(e))
room['turnIndex']=0
srv.call_dice(room['code'],'p-host',8,2)
print('break ok', room['currentCall'])