"""狼人杀多人房间:密钥建房 → 邀请码加入 → 多真人 + AI 补位 → 大厅→对局→是否继续 → 闲置自毁。 隔离:每个 Room 持有自己的一局(引擎 + 后台线程 + 多真人控制器),房间之间互不干扰。 身份:每个真人有 token(浏览器存),据此找房+座位。退出/掉线>60s → AI 补位;持 token 回来 poll → 夺回原座位。 密钥:GENESIS_WW_ROOM_KEY(.env,不入库)校验"能否建房";邀请码校验"能否加入某房"。 生命周期:LOBBY(房主集合)→ PLAYING → ENDED(房主决定是否继续)→ 新局 / CLOSED(解散)。 """ from __future__ import annotations import functools import os import random import threading import time import uuid from dataclasses import dataclass, field from pathlib import Path from typing import Optional from genesis.obs.logging_setup import get_logger from genesis.werewolf.engine import RandomProvider, WerewolfEngine from genesis.werewolf.human import MultiHumanProvider, WebHumanController logger = get_logger("werewolf.rooms") ROOT = Path(__file__).resolve().parents[3] WORLD = ROOT / "worlds" / "werewolf" LOBBY, PLAYING, ENDED, CLOSED = "lobby", "playing", "ended", "closed" IDLE_TTL = 3600.0 # 1h 无活动自毁(有真人在线但长期挂机的兜底) EMPTY_TTL = 90.0 # 真人全离线超此值 → 结束对局并关房(避免无人却空跑 AI / 残余房间) DISCONNECT_GRACE = 60.0 # 真人无 poll 超过此值 → AI 补位(夺回随时可,只要持 token 回来) TURN_TIMEOUT = 90.0 # 每回合真人时限(公示),到点 AI 托管该次决策 def _env(name: str) -> str: """读环境变量,缺失则回退项目根 .env(密钥等不入库)。""" v = os.getenv(name) if v: return v envf = ROOT / ".env" if envf.exists(): for ln in envf.read_text(encoding="utf-8").splitlines(): ln = ln.strip() if ln.startswith(name + "="): return ln.split("=", 1)[1] return "" @functools.lru_cache(maxsize=1024) def tts_mp3(text: str, voice: str) -> bytes: """合成并缓存(相同文本+音色不重复调火山);Room 预合成与 server /api/tts 共用同一缓存。""" from genesis.voice.volc_tts import synthesize return synthesize(text, voice=voice) @dataclass class Player: token: str # 私密身份(浏览器持有,绝不下发给别人) nickname: str pid: str = "" # 公开短 id:房主踢人时定位用(非密钥,可下发) seat: Optional[int] = None # 开局入座后才有 avatar: int = 0 # 头像编号 1-9(入房即分配,大厅+对局一路跟随该玩家) last_seen: float = 0.0 is_creator: bool = False class Room: """一个隔离房间 = 一局(可连续多局)的容器:大厅 + 引擎 + 多真人控制器 + 生命周期。""" def __init__(self, room_id: str, invite_code: str, *, brain: str = "llm", board: str = "9_swh") -> None: self.id = room_id self.invite_code = invite_code self.creator_token: Optional[str] = None self.brain = brain self.board_id = board self.phase = LOBBY self.players: dict[str, Player] = {} self.lock = threading.RLock() self.last_activity = time.monotonic() # 对局态(开局时建) self.engine: Optional[WerewolfEngine] = None self.provider: Optional[MultiHumanProvider] = None self.controllers: dict[int, WebHumanController] = {} self.kicked: set = set() # 被房主踢出的 token(其 poll 会收到 phase:kicked → 前端回落地页) self.seat_avatar: dict[int, int] = {} # 座位→头像编号(开局定:真人沿用自己的,AI 取剩余) self._thread: Optional[threading.Thread] = None self.live: Optional[dict] = None self.voice_on = True self._spoken = threading.Event() self._spoken_want: Optional[str] = None # 当前正在等待"播完"的那段发言 key(keyed 节拍,治末位被切/轮次重叠) self.winner: Optional[str] = None self.game_id = 0 # 每开一局 +1:前端据此重置上一局的去重/动效状态 self._ai_models = None # 缓存 (fast, think) 模型,跨局复用,免重复构建 def touch(self) -> None: self.last_activity = time.monotonic() @property def aborted(self) -> bool: """本局是否已中止(房关 / 引擎被置 aborted)。语音节拍据此立即放行,不空等。 ★必须存在:_wait_audio 引用 self.aborted,缺它会抛 AttributeError 被引擎 try/except 吞掉, 导致每段发言的『等播完』钩子静默失效 → 引擎抢跑、文字全冒、语音错位、乱序。""" return self.phase == CLOSED or bool(self.engine and self.engine.aborted) def capacity(self) -> int: return 9 # 9 人板;后续按 board.seats 取 def human_count(self) -> int: with self.lock: return len(self.players) # ───────── 大厅 / 成员 ───────── def _alloc_avatar(self) -> int: used = {p.avatar for p in self.players.values()} for a in range(1, self.capacity() + 1): if a not in used: return a return 1 def add_player(self, nickname: str, creator: bool = False) -> Player: with self.lock: token = uuid.uuid4().hex nick = (nickname or "").strip()[:16] or ("房主" if creator else f"玩家{len(self.players) + 1}") p = Player(token=token, nickname=nick, pid=uuid.uuid4().hex[:8], avatar=self._alloc_avatar(), is_creator=creator, last_seen=time.monotonic()) self.players[token] = p if creator and not self.creator_token: self.creator_token = token self.touch() return p def can_join(self) -> bool: with self.lock: return self.phase in (LOBBY, ENDED) and len(self.players) < self.capacity() def online_count(self) -> int: now = time.monotonic() with self.lock: return sum(1 for p in self.players.values() if (now - p.last_seen) < DISCONNECT_GRACE) def _roster(self) -> list: now = time.monotonic() with self.lock: return [{"seat": p.seat, "nickname": p.nickname, "pid": p.pid, "avatar": p.avatar, "is_creator": p.is_creator, "online": (now - p.last_seen) < DISCONNECT_GRACE} for p in sorted(self.players.values(), key=lambda x: (x.seat or x.avatar or 99))] # ───────── 对局 ───────── def _build_ai(self, eng: WerewolfEngine, rng: random.Random): if self.brain == "llm": from genesis.llm.lc_adapter import build_chat_model from genesis.werewolf.brain import MindProvider, load_personas # 只用快模型:实测 thinking 模式做结构化发言会被推理 token 耗尽预算→空函数调用→全兜底"N号发言"且每句~90s; # 快模型 ~5s 出稳定连贯发言。故不再注入 make_think_model(speak/digest 的 think=True 自动回落到快模型)。 if self._ai_models is None: self._ai_models = build_chat_model(temperature=0.8, max_tokens=480, request_timeout=60) return MindProvider.for_game(eng, make_model=lambda **kw: self._ai_models, personas=load_personas(WORLD)) # 座位人设:让 9 个 AI 各有其味 return RandomProvider(rng) def start_game(self, by_token: str) -> bool: """房主开局(LOBBY 或 ENDED→新一局):发牌 + 在房真人随机入座 + AI 补满 + 起对局线程。""" with self.lock: if by_token != self.creator_token or self.phase not in (LOBBY, ENDED): return False rng = random.Random() eng = WerewolfEngine(WORLD, self.board_id, RandomProvider(rng), rng=rng) eng.pace = 3.0 eng.think_pause = 10.0 # 每位发言后全场消化停顿(秒);原 20s 偏长,10s 体验更紧凑 eng.wait_audio = self._wait_audio eng.prewarm_tts = self._prewarm prov = MultiHumanProvider(self._build_ai(eng, rng), eng.roles) eng.is_human = prov.has_human # 预生成发言时跳过当前真人座位 eng.provider = prov self.controllers = {} seats = list(eng.state.roles) rng.shuffle(seats) # 真人随机入座 for p, st in zip(list(self.players.values()), seats): p.seat = st c = WebHumanController(timeout=TURN_TIMEOUT) self.controllers[st] = c prov.attach(st, c) # 座位→头像:真人沿用自己入房时的头像,AI 座位取剩余头像(全 9 个不重复) self.seat_avatar = {} used = set() for p in self.players.values(): if p.seat is not None: self.seat_avatar[p.seat] = p.avatar; used.add(p.avatar) pool = [a for a in range(1, 10) if a not in used] for st in eng.state.roles: if st not in self.seat_avatar: self.seat_avatar[st] = pool.pop(0) if pool else st self.engine, self.provider = eng, prov self.winner = None self.live = None self.game_id += 1 self.phase = PLAYING self.touch() self._thread = threading.Thread(target=self._run, daemon=True, name=f"room-{self.id}") self._thread.start() logger.info("房间 %s 开局:真人 %d 人入座 %s,AI 补满,brain=%s", self.id, len(self.players), [p.seat for p in self.players.values()], self.brain) return True def _run(self) -> None: try: self.engine.run_game() except Exception: logger.exception("房间 %s 对局线程崩溃", self.id) finally: with self.lock: self.winner = self.engine.state.winner if self.engine else None if self.phase == PLAYING: # 正常打完 → 等房主决定是否继续(被 close 则保持 CLOSED) self.phase = ENDED self.touch() def continue_game(self, by_token: str, cont: bool) -> bool: with self.lock: if by_token != self.creator_token or self.phase != ENDED: return False if cont: return self.start_game(by_token) # 同房玩家,重新发牌开下一局 self.close(by_token) return True def close(self, by_token: str) -> bool: """房主关房:置 CLOSED + 中止对局 + 唤醒所有等待中的真人(对局线程随即收尾)。""" with self.lock: if by_token != self.creator_token: return False self.phase = CLOSED if self.engine: self.engine.aborted = True for c in self.controllers.values(): c.release() self._spoken.set() self.touch() return True def force_destroy(self) -> None: """管理器清理调用:中止对局线程(daemon,随后自然退出)。""" with self.lock: self.phase = CLOSED if self.engine: self.engine.aborted = True for c in self.controllers.values(): c.release() self._spoken.set() # ───────── 真人在环中(掉线补位 / 夺回) ───────── def _maybe_autoclose(self) -> None: """一个真人都不剩了(全退/全被踢)→ 立即结束对局 + 关房(不空跑 AI;管理器随后销毁)。""" if not self.players and self.phase != CLOSED: self.phase = CLOSED if self.engine: self.engine.aborted = True for c in self.controllers.values(): c.release() self._spoken.set() logger.info("房间 %s:无真实玩家,自动结束对局并关闭", self.id) def leave(self, token: str) -> None: """显式退出:让出座位 → AI 补位;移除玩家(再回来是新身份,夺不回原座位,符合"显式退出")。""" with self.lock: p = self.players.pop(token, None) if p and p.seat is not None and self.provider: self.provider.detach(p.seat) self._maybe_autoclose() self.touch() def kick(self, by_token: str, target_pid: str) -> bool: """房主踢人:按公开 pid 定位(不能踢自己);移出房间 + 座位转 AI;该玩家下次 poll 收到 phase:kicked。""" with self.lock: if by_token != self.creator_token: return False target = next((p for p in self.players.values() if p.pid == target_pid), None) if not target or target.token == self.creator_token: return False self.players.pop(target.token, None) self.kicked.add(target.token) if target.seat is not None and self.provider: self.provider.detach(target.seat) # AI 补位 self._maybe_autoclose() # 踢到一个不剩 → 关房 self.touch() logger.info("房间 %s:房主踢出 %s(%s)", self.id, target.nickname, target_pid) return True def check_disconnects(self) -> None: """掉线>60s 的真人座位 → AI 补位(持 token 回来 poll 会自动夺回)。管理器周期调用。""" if self.phase != PLAYING or not self.provider: return now = time.monotonic() with self.lock: for p in self.players.values(): if (p.seat is not None and self.provider.has_human(p.seat) and (now - p.last_seen) > DISCONNECT_GRACE): self.provider.detach(p.seat) logger.info("房间 %s:%d号(%s)掉线>%.0fs → AI 补位", self.id, p.seat, p.nickname, DISCONNECT_GRACE) def _reclaim_if_needed(self, p: Player) -> None: """玩家 poll 时:若其座位正被 AI 接管且本人还活着 → 夺回(重新挂控制器)。""" if (self.phase == PLAYING and p.seat is not None and self.provider and self.engine and not self.provider.has_human(p.seat) and self.engine.state.is_alive(p.seat)): c = self.controllers.get(p.seat) or WebHumanController(timeout=TURN_TIMEOUT) self.controllers[p.seat] = c self.provider.attach(p.seat, c) logger.info("房间 %s:%d号(%s)回归 → 夺回座位", self.id, p.seat, p.nickname) # ───────── 语音节拍(引擎钩子)───────── def _wait_audio(self, seat: int, text: str) -> None: """阻塞到『这一段发言』在前端真正播完(keyed,避免上一段的 spoken 误满足本段→末位被切/轮次重叠 bug6/bug7)。 无在场真人/无人回报则超时兜底。关语音/房关则不等。""" if not self.voice_on or self.phase == CLOSED or self.aborted: return key = self._spoken_key(seat, text) with self.lock: self._spoken_want = key # 只认本段的播完回报 self._spoken.clear() est = min(40.0, max(4.0, len(text or "") / 4.5)) deadline = time.monotonic() + est + 12.0 while time.monotonic() < deadline: if self.phase == CLOSED or self.aborted: return if self._spoken.wait(timeout=0.5): # 收到匹配本段的 spoken → 立即返回 return with self.lock: self._spoken_want = None # 超时兜底,清掉期望 @staticmethod def _spoken_key(seat, text) -> str: return f"{seat}|{(text or '')[:48]}" def _prewarm(self, seat: int, text: str) -> None: if not self.voice_on or not text: return from genesis.voice.volc_tts import voice_for_seat try: tts_mp3(text.strip()[:400], voice_for_seat(seat)) except Exception: pass def mark_spoken(self, seat=None, text=None) -> None: """前端报『某段发言播完』。只有匹配当前正在等的那段(_spoken_want)才唤醒引擎—— 防止前一段的播完信号误满足后一段的等待(致末位被切/轮次重叠)。无 seat(旧前端)则兜底直接唤醒。""" with self.lock: want = self._spoken_want if want is None: return if seat is None or self._spoken_key(int(seat) if str(seat).lstrip('-').isdigit() else seat, text) == want: self._spoken_want = None self._spoken.set() def sink(self, seat: int, partial: str) -> None: self.live = {"seat": seat, "text": partial[:500]} # ───────── 给前端的一帧 ───────── def state_for(self, token: str) -> dict: with self.lock: if token in self.kicked: # 被踢:告知前端 → 清 token 回落地页 return {"phase": "kicked"} p = self.players.get(token) if p: p.last_seen = time.monotonic() self._reclaim_if_needed(p) self.touch() seat = p.seat if p else None frame = self.engine.frame(seat) if (self.engine and self.phase != LOBBY) else None awaiting = None if p and p.seat is not None and self.phase == PLAYING and self.provider and self.provider.has_human(p.seat): c = self.controllers.get(p.seat) awaiting = c.pending() if c else None return { "phase": self.phase, "room": self.id, "invite": self.invite_code, "game_id": self.game_id, "you": ({"seat": seat, "nickname": p.nickname, "is_creator": p.is_creator} if p else None), "players": self._roster(), "seat_avatar": dict(self.seat_avatar), "human_count": len(self.players), "capacity": self.capacity(), "frame": frame, "awaiting": awaiting, "live": None if awaiting else self.live, "winner": self.winner, "running": bool(self._thread and self._thread.is_alive()), } def submit(self, token: str, answer) -> bool: with self.lock: p = self.players.get(token) if not p or p.seat is None: return False c = self.controllers.get(p.seat) self.touch() return c.submit(answer) if c else False class RoomManager: """全局房间表:建房(密钥)/加入(邀请码)/按 token 找房 + 后台清理。线程安全。""" def __init__(self, brain: str = "llm") -> None: self.brain = brain self.rooms: dict[str, Room] = {} self.by_invite: dict[str, str] = {} self.by_token: dict[str, str] = {} self.lock = threading.RLock() def create(self, key: str, nickname: str = "") -> Optional[dict]: if not key or key != _env("GENESIS_WW_ROOM_KEY"): # 密钥校验:能否建房 return None with self.lock: rid = uuid.uuid4().hex[:8] code = self._gen_code() room = Room(rid, code, brain=self.brain) creator = room.add_player(nickname, creator=True) self.rooms[rid] = room self.by_invite[code] = rid self.by_token[creator.token] = rid logger.info("建房 %s 邀请码 %s", rid, code) return {"room_id": rid, "invite_code": code, "token": creator.token} def join(self, code: str, nickname: str = "") -> dict: with self.lock: rid = self.by_invite.get((code or "").strip().upper()) room = self.rooms.get(rid) if rid else None if not room: return {"error": "房间不存在或邀请码无效"} if not room.can_join(): return {"error": "房间已满或对局进行中"} p = room.add_player(nickname) self.by_token[p.token] = rid return {"room_id": rid, "token": p.token, "invite_code": room.invite_code} def room_of(self, token: str) -> Optional[Room]: with self.lock: rid = self.by_token.get(token) return self.rooms.get(rid) if rid else None def _gen_code(self) -> str: alpha = "ABCDEFGHJKLMNPQRSTUVWXYZ23456789" # 去掉易混字符 while True: code = "".join(random.choice(alpha) for _ in range(6)) if code not in self.by_invite: return code def cleanup(self) -> None: """周期清理:已关/无真人 → 销毁;真人全离线超 EMPTY_TTL → 结束对局+关房;否则查掉线补位。""" now = time.monotonic() with self.lock: dead = [] for rid, room in self.rooms.items(): idle = now - room.last_activity if room.phase == CLOSED or room.human_count() == 0: # 已关 / 一个真人都没有 → 销毁 dead.append(rid) elif room.online_count() == 0 and idle > EMPTY_TTL: # 真人全离线超时 → 结束对局并关房 dead.append(rid) elif idle > IDLE_TTL: # 兜底:超长无活动 dead.append(rid) else: room.check_disconnects() for rid in dead: self._destroy(rid) # force_destroy 会中止对局线程 def _destroy(self, rid: str) -> None: room = self.rooms.pop(rid, None) if not room: return room.force_destroy() self.by_invite.pop(room.invite_code, None) for t in [t for t, r in list(self.by_token.items()) if r == rid]: self.by_token.pop(t, None) logger.info("房间 %s 已销毁", rid)