"""WerewolfEngine:「阶段 + 能力词表」状态机。 夜间按 night_order 排序、按 role.night_action 分发(kill/check/save_poison/…)、按 counts_as 判屠边 —— 全程**不写 "if 角色==预言家"**。决策来自可插拔的 DecisionProvider: - M0 = RandomProvider(随机合法决策,跑通流程,无 LLM); - M1 = AI 脑(实现同一接口);M2 = 真人到点 await(在 provider 之上加等待层)。 """ from __future__ import annotations import random import threading import time from collections import Counter from pathlib import Path from typing import Optional, Protocol from genesis.obs.logging_setup import get_logger from genesis.werewolf import perception from genesis.werewolf.board import Board, deal, load_board, resolve_night_order from genesis.werewolf.roles import Role, load_roles from genesis.werewolf import state as S from genesis.werewolf.state import GameState logger = get_logger("werewolf.engine") MAX_ROUNDS = 30 # 安全阀:正常对局远早于此分胜负,纯防御异常死循环 # 阶段中文标签(前端展示用) PHASE_LABEL = { S.LOBBY: "等待开局", S.SHERIFF_RUN: "警长竞选", S.SHERIFF_SPEECH: "上警发言", S.SHERIFF_VOTE: "警徽投票", S.NIGHT: "天黑请闭眼", S.DAY_ANNOUNCE: "天亮了", S.DAY_SPEECH: "轮流发言", S.DAY_VOTE: "放逐投票", S.DAY_EXILE: "投票结算", S.END: "对局结束", } class DecisionProvider(Protocol): """引擎在每个决策点回调它;引擎已算好合法候选,provider 只需在其中做选择。""" def wolf_discuss(self, state: GameState, seat: int, candidates: list[int]) -> dict: ... # {target, message},夜间狼私聊 def decide_kill(self, state: GameState, seat: int, candidates: list[int]) -> Optional[int]: ... def decide_check(self, state: GameState, seat: int, candidates: list[int]) -> Optional[int]: ... def decide_witch(self, state: GameState, seat: int, knife: Optional[int], can_save: bool, can_poison: bool, poison_candidates: list[int]) -> dict: ... def decide_run_sheriff(self, state: GameState, seat: int) -> bool: ... def decide_speech_dir(self, state: GameState, seat: int) -> dict: ... # 警长定序:{direction:'cw'|'ccw'};可选,引擎用 getattr 兜底 def decide_withdraw(self, state: GameState, seat: int) -> bool: ... # 上警退水;可选 def decide_badge_pass(self, state: GameState, seat: int, candidates: list[int]) -> Optional[int]: ... # 警长临终移交警徽,None=撕毁;可选 def decide_explode(self, state: GameState, seat: int) -> bool: ... # 狼人白天自爆;可选 def speak(self, state: GameState, seat: int) -> str: ... def decide_vote(self, state: GameState, seat: int, candidates: list[int]) -> Optional[int]: ... def decide_hunter_shoot(self, state: GameState, seat: int, candidates: list[int]) -> Optional[int]: ... def digest_day(self, state: GameState, seat: int) -> None: ... # 每天收尾的记忆压缩(可 no-op) class WerewolfEngine: def __init__(self, world_dir: str | Path, board_id: str, provider: DecisionProvider, rng: Optional[random.Random] = None) -> None: world_dir = Path(world_dir) self.roles: dict[str, Role] = load_roles(world_dir / "roles") self.board: Board = load_board(world_dir / "boards" / f"{board_id}.json") self.provider = provider self.rng = rng or random.Random() self._night_order = resolve_night_order(self.board, self.roles) seed = self.rng.randint(0, 2**31 - 1) seats = deal(self.board, self.rng) self.state = GameState(board_id=self.board.id, seats=self.board.seats, roles=seats, seed=seed) for s in self.state.roles: # 按能力初始化药剂(不写 "if 女巫") if self.roles[self.state.role_of(s)].night_action == "save_poison": self.state.witch_potions[s] = {"antidote": True, "poison": True} self.pace = 0.0 # 戏剧停顿(秒):web 会话设>0,让死讯/放逐宣布有时间被看清;测试保持 0 self.think_pause = 0.0 # 每位发言后的思考停顿(秒):web 设 20,给全场看清+思考;测试 0 self.banner: Optional[dict] = None # 当前进度横幅(前端展示"现在在等谁/在干嘛"),由各阶段实时更新 self.wait_audio = None # 可注入 (seat,text)->None:阻塞到该发言语音播完(前端报信),关语音/自己则立即返回 self.prewarm_tts = None # 可注入 (seat,text)->None:提前合成该段语音暖缓存(预生成下一位时一并做) self.is_human = None # 可注入 (seat)->bool:该座位当前是否真人(预生成跳过真人;多真人房动态变化) self._spipe: dict = {} # 发言预生成流水线:seat -> {thread, holder} self.aborted = False # 房间关闭时置 True:对局在阶段间尽快收尾,不空跑 def _beat(self, mult: float = 1.0) -> None: """在关键转折处停一拍(天亮宣布、放逐结算),给前端宣布动画时间;pace=0 时不停(测试)。""" if self.pace > 0: time.sleep(self.pace * mult) # —— 进度横幅:任何"要等一会儿"的地方都先点亮它,玩家才知道在等谁/在干嘛(治"傻等不知进度")—— def _set_banner(self, kind: str, text: str, seat: Optional[int] = None, secs: float = 0.0) -> None: self.banner = {"kind": kind, "text": text, "seat": seat, "until": (time.time() + secs) if secs > 0 else None} def _clear_banner(self) -> None: self.banner = None def _hold(self, kind: str, text: str, seat: Optional[int], secs: float) -> None: """点亮带倒计时的横幅并真正停住引擎线程 secs 秒(pace>0 才停);前端据此显示倒计时、并行播放语音。""" if self.pace <= 0 or secs <= 0 or self.aborted: # 已关闭则不再停顿,尽快收尾 return self._set_banner(kind, text, seat, secs) time.sleep(secs) def _after_speech(self, seat: int, text: str) -> None: """一位发言彻底结束的节拍:先等其语音真正播完(前端 audio 结束回报,而非估时),再进入思考停顿。 语音没播完不进思考;关语音则不空等。这保证『说完→听完→才思考→才下一位』。 ★真人本人的发言不等播音:他是自己说出来的(前端不回放自己的发言、也不报 spoken), 再去等 audio 信号只会傻等到超时——点发送应当立刻进入思考倒计时。其它真人客户端会在思考间隙听到他的 TTS。""" speaker_is_human = bool(self.is_human and self.is_human(seat)) if self.wait_audio is not None and not speaker_is_human: self._set_banner("speaking", f"{seat}号 发言中…", seat) # 无倒计时:时长由真实语音决定 try: self.wait_audio(seat, text) # 阻塞到前端报"播完"(或会话内置超时兜底) except Exception: pass if self.think_pause > 0: self._hold("thinking", "思考时间 · 消化上一位发言…", seat, self.think_pause) def _narrate(self, text: str, etype: str = "judge") -> None: """法官播报:记一条法官台词(进 feed)+ 点亮法官横幅 + 等其语音播完再继续(用语音引领节奏)。 seat=0 代表法官;关语音/无前端则不空等(wait_audio 立即返回/超时兜底)。""" self.state.record(etype, text) # seat=None → 法官 self._set_banner("judge", "🎙 法官 · " + text) if self.prewarm_tts is not None: # 预合成法官语音(暖缓存),前端取时即时 try: self.prewarm_tts(0, text) except Exception: pass if self.wait_audio is not None: try: self.wait_audio(0, text) except Exception: pass # —— 并发决策:顺序无关的决策(上警/投票)并行跑,提速且互不可见 —— def _parallel(self, items: list, fn) -> dict: """对每个 item 并发跑 fn(item) → {item: result}。AI 多线程提速;真人那一支在自己线程里等 UI。 并发上限不在这里限,而在 LLM 调用层(brain._LLM_SEM)——这样真人等 UI 不占 LLM 名额。""" out: dict = {} def worker(it): try: out[it] = fn(it) except Exception as e: logger.warning("[parallel] %s 失败:%s", it, str(e)[:80]); out[it] = None ths = [threading.Thread(target=worker, args=(it,), daemon=True) for it in items] for t in ths: t.start() for t in ths: t.join() return out def _collect_votes(self, voters: list, cands_fn, kind: str) -> dict: """并发收票:期间**不写 votes_log**(谁都看不到本轮票 → 杜绝跟票),全收齐后统一公布。返回 {voter: target}。""" self._set_banner("waiting", "全场同时投票中…(互相看不到票)") results = self._parallel(voters, lambda v: self.provider.decide_vote(self.state, v, cands_fn(v))) for v in voters: # 此刻才公布到 votes_log t = results.get(v) if t in cands_fn(v): self.state.votes_log.append({"day": self.state.day, "kind": kind, "voter": v, "target": t}) return results def _announce_votes(self, results: dict, label: str) -> None: """票型公布(bug5):全场可见谁投了谁——进 feed + 法官语音播报。""" arrows = "，".join(f"{v}号→{t}号" for v, t in sorted(results.items()) if t is not None) self._narrate(f"{label}票型:{arrows}。" if arrows else f"{label}无人投票。", "announce") # —— 发言预生成流水线:上一位『揭晓后』就并行算下一位(文字+语音),轮到才放出 —— def _prefetch_speech(self, seat: int) -> None: if (self.is_human and self.is_human(seat)) or seat in self._spipe: # 真人不预取(等其输入) return holder: dict = {} def work(): try: holder["text"] = self.provider.speak(self.state, seat) # 非流式(未设 stream_sink) if self.prewarm_tts and holder.get("text"): self.prewarm_tts(seat, holder["text"]) # 顺便预合成语音 except Exception as e: logger.warning("预生成 %d 号发言失败:%s", seat, str(e)[:100]) th = threading.Thread(target=work, daemon=True); th.start() self._spipe[seat] = {"thread": th, "holder": holder} def _take_speech(self, seat: int) -> str: p = self._spipe.pop(seat, None) human_now = bool(self.is_human and self.is_human(seat)) # 预生成后若被真人夺回 → 弃用预生成,改问真人 if p and not human_now: p["thread"].join() # 等预生成线程结束(有界:模型超时×重试封顶),避免重复发起调用 if p["holder"].get("text") is not None: return p["holder"]["text"] # 预生成结果(真发言;全失败时已是兜底文案) return self.provider.speak(self.state, seat) # 真人(阻塞等输入)或未预取时现场生成 def _run_speeches(self, order: list[int], etype: str, allow_explode: bool = False) -> bool: """逐位发言:下一位在上一位 record 后即并行预生成,轮到时直接揭晓(文字+语音都在自己轮次)。 allow_explode=True(白天发言)时,轮到狼可选择自爆→中止当天发言。返回是否发生了自爆。 aborted(房间关闭/分胜负)立即收尾,不再空跑发言(治 bug10:结束后还出发言)。""" if order: self._prefetch_speech(order[0]) for i, seat in enumerate(order): if self.aborted or self.state.winner: # 已收尾/已分胜负 → 立即停,别再产出发言 return False if allow_explode and self._role(seat).team == "wolf" and self._ask_explode(seat): self._spipe.pop(seat, None) # 丢弃其预生成发言 self.state.kill(seat, S.CAUSE_EXPLODE) self.state.record("exile", f"{seat}号狼人自爆,掀桌亮明身份!白天中止,直接入夜。", seat=seat) self._narrate(f"{seat}号自爆出局,白天发言与投票中止,即刻天黑。", "exile") self._transfer_badge_if_dead() # 自爆者恰是警长 → 移交警徽 return True self._set_banner("speaking", f"{seat}号 发言中…", seat=seat) text = self._take_speech(seat) self.state.record(etype, text, seat=seat) # 揭晓 → feed,前端据此放语音 if i + 1 < len(order): self._prefetch_speech(order[i + 1]) # 本位已 record,下一位看得到 → 预生成(在下面的等待间隙完成) self._after_speech(seat, text) return False def _ask_explode(self, seat: int) -> bool: fn = getattr(self.provider, "decide_explode", None) # 可选接口,旧 provider 缺它=不自爆 try: return bool(fn(self.state, seat)) if fn else False except Exception: return False # —— 阵营/归边(引擎持有 Role 注册表,故由它做)—— def _role(self, seat: int) -> Role: return self.roles[self.state.role_of(seat)] def _alive(self, pred) -> list[int]: return [s for s in self.state.alive_seats() if pred(s)] def _wolves_alive(self) -> list[int]: return self._alive(lambda s: self._role(s).team == "wolf") def _gods_alive(self) -> list[int]: return self._alive(lambda s: self._role(s).counts_as == "god") def _villagers_alive(self) -> list[int]: return self._alive(lambda s: self._role(s).counts_as == "villager") # ───────────────── 主循环 ───────────────── def run_game(self) -> Optional[str]: self.state.day = 1 for _ in range(MAX_ROUNDS): if self.aborted: break self._night() if self.aborted or self._check_win(): break self._day(first=(self.state.day == 1)) if self.aborted or self._check_win(): break self._digest_day() # 每天收尾:各存活 agent 跑一次记忆压缩(脑层钩子) self.state.day += 1 self._clear_banner() logger.info("[狼人杀] 对局结束:%s 胜(%d 天)", self.state.winner or "无", self.state.day) return self.state.winner # ───────────────── 渲染快照(前端消费)───────────────── def frame(self, viewer: Optional[int] = None) -> dict: """一帧渲染态。 - viewer=None 或已出局 → **上帝视角**(揭示全部真名 + 全量 transcript 含狼私聊); - 活着的 viewer → **玩家视角**:只亮自己身份(狼额外亮队友),按 perception 给 transcript, 并附 private 块(自己的查验史/刀型/药/队友)供侧栏显示。""" god = viewer is None or not self.state.is_alive(viewer) or bool(self.state.winner) # 终局全员揭晓(bug11) pv = None if god else perception.view(self.state, viewer, self.roles) teammates = set(pv.teammates) if (pv and pv.teammates) else set() seats = [] for s in sorted(self.state.roles): r = self.roles[self.state.role_of(s)] reveal = god or (viewer is not None and (s == viewer or s in teammates)) seats.append({ "seat": s, "alive": self.state.is_alive(s), "is_sheriff": s == self.state.sheriff, "role": r.name if reveal else None, "team": r.team if reveal else None, }) if god: tx, deaths = list(self.state.transcript[-24:]), list(self.state.deaths) else: tx, deaths = list(pv.transcript[-24:]), list(pv.dead) private = None if pv: private = {"role": pv.role_name, "team": pv.team, "teammates": [s for s in (pv.teammates or []) if s != viewer] or None, "checks": pv.check_results, "knife": pv.knife_target, "potions": pv.potions} banner = None if self.banner: # 实时算剩余秒(服务端为准),前端只管显示 b = self.banner left = max(0, int(round(b["until"] - time.time()))) if b.get("until") else None banner = {"kind": b["kind"], "text": b["text"], "seat": b.get("seat"), "secs_left": left} return { "board": self.board.name, "day": self.state.day, "phase": self.state.phase, "phase_label": PHASE_LABEL.get(self.state.phase, ""), "is_night": self.state.phase == S.NIGHT, "sheriff": self.state.sheriff, "winner": self.state.winner, "god": god, "viewer": viewer, "seats": seats, "transcript": tx, "deaths": deaths, "private": private, "banner": banner, } # ───────────────── 夜晚:按能力词表分发 ───────────────── def _night(self) -> None: self.state.phase = S.NIGHT self.state.last_knife = None knife: Optional[int] = None saved: set[int] = set() poison_target: Optional[int] = None self._narrate(f"第 {self.state.day} 天,天黑请闭眼。") # 法官:开夜 NIGHT_BANNER = {"kill": "天黑了 · 狼人睁眼行动中…", "check": "天黑了 · 预言家查验中…", "save_poison": "天黑了 · 女巫抉择中…"} NIGHT_NARRATE = {"kill": "狼人请睁眼,今晚你们要猎杀谁?", "check": "预言家请睁眼,你要查验谁?", "save_poison": "女巫请睁眼,今晚是否用药?"} for role_id in self._night_order: action = self.roles[role_id].night_action actors = self._alive(lambda s: self.state.role_of(s) == role_id) if not actors: continue self._set_banner("night", NIGHT_BANNER.get(action, "天黑了 · 夜晚行动中…")) if action in NIGHT_NARRATE: self._narrate(NIGHT_NARRATE[action]) # 法官:逐个唤起夜间角色(经典主持词) if action == "kill": # 狼队夜间会商:私聊频道里轮流提议→取多数 cands = self._alive(lambda s: self._role(s).team != "wolf") if cands: proposals = [] for w in actors: # 按座位序发言:后发的狼看得到先发的提议(真交流) d = self.provider.wolf_discuss(self.state, w, cands) or {} t = d.get("target") self.state.record("wolf_chat", f"提议刀 {t} 号 —— {d.get('message', '')}".strip(), seat=w, channel="wolf") # 仅狼可见 if t in cands: proposals.append(t) if proposals: knife = Counter(proposals).most_common(1)[0][0] self.state.last_knife = knife # 刀型:仅女巫可见 elif action == "check": # 预言家查验 for seer in actors: cands = [s for s in self.state.alive_seats() if s != seer] t = self.provider.decide_check(self.state, seer, cands) if t in cands: res = "wolf" if self._role(t).team == "wolf" else "good" self.state.seer_checks.setdefault(seer, []).append({"target": t, "result": res}) elif action == "save_poison": # 女巫救/毒 for witch in actors: pot = self.state.witch_potions.get(witch, {"antidote": False, "poison": False}) self_save_ok = self.board.options.get("witch_self_save", False) can_save = pot["antidote"] and knife is not None and (knife != witch or self_save_ok) can_poison = pot["poison"] pcands = [s for s in self.state.alive_seats() if s != witch] d = self.provider.decide_witch(self.state, witch, knife, can_save, can_poison, pcands) if can_save and d.get("save"): saved.add(knife); pot["antidote"] = False both_ok = self.board.options.get("witch_save_and_poison_same_night", True) if can_poison and d.get("poison") in pcands and (both_ok or not d.get("save")): poison_target = d["poison"]; pot["poison"] = False self._set_banner("night", "天亮结算中…") # 结算夜间死亡(守卫等其它能力按词表在此扩展) deaths: list[tuple[int, str]] = [] if knife is not None and knife not in saved: deaths.append((knife, S.CAUSE_KNIFE)) if poison_target is not None: deaths.append((poison_target, S.CAUSE_POISON)) for seat, cause in deaths: self.state.kill(seat, cause) for seat, cause in deaths: # 被刀的猎人可开枪;被毒不能 self._maybe_hunter_shoot(seat, cause) def _digest_day(self) -> None: """脑层钩子:每天收尾触发各存活 agent 的记忆压缩。 ★后台 fire-and-forget(不 barrier 等待):压缩是 LLM 调用,若卡在这里等齐会让"下一阶段发言前傻等很久"(bug8); 改为后台线程跑,自然在随后的黑夜阶段(本就要花时间)完成,不阻塞推进。RandomProvider 为 no-op,开销可忽略。""" for s in list(self.state.alive_seats()): threading.Thread(target=self._safe_digest, args=(s,), daemon=True).start() def _safe_digest(self, seat: int) -> None: try: self.provider.digest_day(self.state, seat) except Exception as e: logger.warning("[digest] %d号 后台压缩失败:%s", seat, str(e)[:80]) def _maybe_hunter_shoot(self, dead_seat: int, cause: str) -> None: role = self.roles[self.state.role_of(dead_seat)] if role.on_death != "hunter_shoot" or cause == S.CAUSE_POISON: return cands = [s for s in self.state.alive_seats() if s != dead_seat] if not cands: return t = self.provider.decide_hunter_shoot(self.state, dead_seat, cands) if t in cands: self.state.kill(t, S.CAUSE_SHOT) self.state.record("shot", f"{dead_seat}号开枪带走了{t}号", seat=dead_seat) self._maybe_hunter_shoot(t, S.CAUSE_SHOT) # 连锁(另一个猎人) # ───────────────── 白天:竞选 / 公布 / 发言 / 投票 / 放逐 ───────────────── def _day(self, first: bool) -> None: if first and self.board.sheriff: self._sheriff_election() self.state.phase = S.DAY_ANNOUNCE overnight = [d["seat"] for d in self.state.deaths if d["day"] == self.state.day and d["cause"] in (S.CAUSE_KNIFE, S.CAUSE_POISON, S.CAUSE_SHOT)] self._narrate("天亮了。昨晚是平安夜,无人死亡。" if not overnight # 法官:开天 + 宣布战况 else "天亮了。昨晚，" + "、".join(f"{s}号" for s in sorted(overnight)) + " 倒在了血泊中。", "announce") self._transfer_badge_if_dead() # 警长昨夜被刀/毒 → 临终移交警徽(bug9) self.state.phase = S.DAY_SPEECH # 顺序发言(警长起):逐位揭晓 + 预生成下一位 self._narrate("现在进入发言环节,请各位依次陈述、各抒己见。") exploded = self._run_speeches(self._speech_order(), "speech", allow_explode=True) self._clear_banner() if not exploded: # 狼自爆 → 中止白天,不再投票,直接入夜 self._day_vote() def _speech_order(self) -> list[int]: """发言序:有警长则由警长定序(选左/右手边起,自己压轴);无警长按座位序。""" alive = self.state.alive_seats() sh = self.state.sheriff if sh not in alive: return alive i = alive.index(sh) others = alive[i + 1:] + alive[:i] # 警长下家起、座位顺位(cw) self._set_banner("waiting", f"{sh}号警长定序中 · 选择发言方向…", seat=sh) fn = getattr(self.provider, "decide_speech_dir", None) # 可选接口:旧 provider 缺它则默认顺位 choice = (fn(self.state, sh) if fn else None) or {} direction = choice.get("direction") if choice.get("direction") in ("cw", "ccw") else "cw" seq = others if direction == "cw" else list(reversed(others)) side = "左手边(座位顺位)" if direction == "cw" else "右手边(座位逆位)" self._narrate(f"{sh}号警长决定:从自己{side}起依次发言,自己压轴。", "sheriff") return seq + [sh] # 警长压轴最后发言 def _sheriff_election(self) -> None: self.state.phase = S.SHERIFF_RUN self._narrate("第一天,先选警长。想竞选警长的玩家,请表态。") # 法官:开启警长竞选 self._set_banner("waiting", "全场同时决定是否上警…") runs = self._parallel(self.state.alive_seats(), # 并发表态(互不可见) lambda s: self.provider.decide_run_sheriff(self.state, s)) candidates = [s for s in self.state.alive_seats() if runs.get(s)] if not candidates: self._narrate("无人竞选警长,警徽流失。", "sheriff") self._clear_banner() return self.state.phase = S.SHERIFF_SPEECH self._narrate("上警玩家:" + "、".join(f"{s}号" for s in candidates) + ",请依次发表竞选宣言。", "sheriff") self._run_speeches(candidates, "sheriff_speech") # 竞选发言:逐位揭晓 + 预生成下一位 # 退水环节(bug1):竞选发言后,上警者可选择退水退出竞选;退水者回到警下可投票 self._set_banner("waiting", "上警玩家抉择是否退水…") wd = self._parallel(candidates, lambda s: self._ask_withdraw(s)) withdrawn = [s for s in candidates if wd.get(s)] candidates = [s for s in candidates if not wd.get(s)] if withdrawn: self._narrate("、".join(f"{s}号" for s in withdrawn) + " 宣布退水,退出竞选。", "sheriff") if not candidates: self._narrate("上警玩家全部退水,警徽流失。", "sheriff") self._clear_banner() return self.state.phase = S.SHERIFF_VOTE # 警下(未上警者 + 退水者)投票 self._narrate("请未上警/已退水的玩家同时投票,选出你心目中的警长。") electorate = [s for s in self.state.alive_seats() if s not in candidates] results = self._collect_votes(electorate, lambda v: list(candidates), "sheriff") # 并发收票 self._announce_votes(results, "警长竞选") # 票型公布(bug5) tally: Counter = Counter(t for t in results.values() if t in candidates) sheriff = self._top_or_none(tally) self.state.sheriff = sheriff self._narrate(f"{sheriff}号当选警长,获得 1.5 票与发言定序权。" if sheriff else "警长竞选平票,警徽流失。", "sheriff") self._clear_banner() def _ask_withdraw(self, seat: int) -> bool: fn = getattr(self.provider, "decide_withdraw", None) # 可选接口,旧 provider 缺它=不退水 try: return bool(fn(self.state, seat)) if fn else False except Exception: return False def _transfer_badge_if_dead(self) -> None: """警长已不在场(被刀/毒/枪/放逐)→ 由倒下的警长移交警徽给一名存活者,或撕毁(bug9)。幂等:警长健在则不动。""" sh = self.state.sheriff if sh is None or self.state.is_alive(sh): return cands = self.state.alive_seats() fn = getattr(self.provider, "decide_badge_pass", None) target = None if fn and cands: try: target = fn(self.state, sh, cands) except Exception: target = None target = target if target in cands else None self.state.sheriff = target self._narrate(f"{sh}号(警长)倒下,临终将警徽移交给 {target}号。" if target else f"{sh}号(警长)倒下,警徽被撕毁,本局不再设警长。", "sheriff") def _day_vote(self) -> None: self.state.phase = S.DAY_VOTE self._narrate("发言结束,请大家同时投票,放逐一名你最怀疑的玩家。") # 法官:进入投票 alive = self.state.alive_seats() cands_fn = lambda v: [s for s in alive if s != v] results = self._collect_votes(alive, cands_fn, "exile") # 并发收票,互不可见 self._announce_votes(results, "放逐投票") # 票型公布(bug5) tally: Counter = Counter() for voter, t in results.items(): if t in cands_fn(voter): tally[t] += self._role(voter).vote_weight * (1.5 if voter == self.state.sheriff else 1.0) exiled = self._top_or_none(tally) self.state.phase = S.DAY_EXILE self._clear_banner() if exiled is None: self._narrate("投票平票,本轮无人出局。", "exile") return self.state.kill(exiled, S.CAUSE_EXILE) self._narrate(f"{exiled}号被投票放逐,请离场。", "exile") # 法官:宣布放逐(语音+横幅,前端另有出局动画) self._maybe_hunter_shoot(exiled, S.CAUSE_EXILE) # 被放逐的猎人可开枪 self._transfer_badge_if_dead() # 警长被放逐(或其枪带走了警长)→ 临终移交警徽(bug9) def _top_or_none(self, tally: Counter) -> Optional[int]: """最高票者;平票按 rng 在并列者中裁决(保证推进,PK 流程留待 M1/M2)。""" if not tally: return None top = max(tally.values()) leaders = [s for s, v in tally.items() if v == top] return self.rng.choice(leaders) # ───────────────── 胜负(按 counts_as/team)───────────────── def _check_win(self) -> bool: if self.state.winner: return True if not self._wolves_alive(): self.state.winner = "good" elif self.board.win == "kill_side" and (not self._gods_alive() or not self._villagers_alive()): self.state.winner = "wolf" elif self.board.win == "kill_all" and not (self._gods_alive() or self._villagers_alive()): self.state.winner = "wolf" if self.state.winner: self.state.phase = S.END self._narrate("游戏结束。" + ("狼人阵营获胜!" if self.state.winner == "wolf" else "狼人全部出局,好人阵营获胜!"), "end") return True return False class RandomProvider: """随机但合法的决策(M0 跑通流程 / 后续真人超时托管的兜底 bot)。给种子即可复现。""" def __init__(self, rng: Optional[random.Random] = None) -> None: self.rng = rng or random.Random() def wolf_discuss(self, state, seat, candidates): return {"target": self.rng.choice(candidates) if candidates else None, "message": "随便挑一个边路的。"} def decide_kill(self, state, seat, candidates): return self.rng.choice(candidates) if candidates else None def decide_check(self, state, seat, candidates): return self.rng.choice(candidates) if candidates else None def decide_witch(self, state, seat, knife, can_save, can_poison, poison_candidates): save = can_save and self.rng.random() < 0.6 poison = (self.rng.choice(poison_candidates) if (can_poison and not save and poison_candidates and self.rng.random() < 0.3) else None) return {"save": save, "poison": poison} def decide_run_sheriff(self, state, seat): return self.rng.random() < 0.5 def decide_speech_dir(self, state, seat): return {"direction": self.rng.choice(["cw", "ccw"])} def decide_withdraw(self, state, seat): return self.rng.random() < 0.12 # 偶尔退水 def decide_badge_pass(self, state, seat, candidates): return self.rng.choice(candidates) if candidates else None def decide_explode(self, state, seat): return False # 随机 bot 不自爆(避免无脑掀桌) def speak(self, state, seat): return f"{seat}号发言。" def decide_vote(self, state, seat, candidates): return self.rng.choice(candidates) if candidates else None # 始终投票,保证推进 def decide_hunter_shoot(self, state, seat, candidates): return self.rng.choice(candidates) if candidates else None def digest_day(self, state, seat): return None # 随机 bot 不做记忆压缩(no-op)