929 lines
40 KiB
Python
929 lines
40 KiB
Python
from __future__ import annotations
|
|
|
|
from copy import deepcopy
|
|
from typing import Any
|
|
|
|
from .models import GameState, utc_now_iso
|
|
|
|
ALLOWED_CLUES = {
|
|
"bloodied_key",
|
|
"torn_manifest",
|
|
"silver_bell",
|
|
"maid_diary",
|
|
"cellar_sigil",
|
|
"missing_guest_note",
|
|
}
|
|
FOGHARBOR_ITEM_LABELS = {
|
|
"bloodied_key": "染血钥匙",
|
|
"torn_manifest": "残破船务清单",
|
|
"silver_bell": "银铃",
|
|
"maid_diary": "女仆日记",
|
|
"cellar_sigil": "地下室印记",
|
|
"missing_guest_note": "失踪者便条",
|
|
}
|
|
FOGHARBOR_ITEM_ALIASES = {
|
|
**{key: key for key in FOGHARBOR_ITEM_LABELS},
|
|
**{value: key for key, value in FOGHARBOR_ITEM_LABELS.items()},
|
|
}
|
|
|
|
NPC_IDS = {"innkeeper", "maid", "silent_traveler"}
|
|
|
|
SCENES = {
|
|
"lobby": {
|
|
"chapter": 1,
|
|
"title": "大厅",
|
|
"description": "壁炉忽明忽暗，旅店老板正用审视的目光打量每个人。",
|
|
"choices": ["检查登记台", "询问旅店老板", "观察沉默旅客"],
|
|
},
|
|
"dining_room": {
|
|
"chapter": 2,
|
|
"title": "餐厅",
|
|
"description": "餐厅里摆着尚有余温的餐具，女仆反复擦拭同一张桌子。",
|
|
"choices": ["检查餐桌残渣", "说服女仆开口", "监听老板与旅客对话"],
|
|
},
|
|
"corridor": {
|
|
"chapter": 3,
|
|
"title": "走廊",
|
|
"description": "长走廊上挂满褪色油画，尽头传来断续的敲击声。",
|
|
"choices": ["追踪敲击声", "查看 301 房门", "搜索暗门机关"],
|
|
},
|
|
"guest_room": {
|
|
"chapter": 4,
|
|
"title": "客房",
|
|
"description": "客房内潮湿阴冷，地板下有若隐若现的符号刻痕。",
|
|
"choices": ["翻找床底箱子", "解读墙上符号", "寻找地下室入口"],
|
|
},
|
|
"basement": {
|
|
"chapter": 5,
|
|
"title": "地下室",
|
|
"description": "地下室中央摆着仪式台，真相与代价都已无处回避。",
|
|
"choices": ["破坏仪式", "与隐藏势力谈判", "尝试组织全员逃离"],
|
|
},
|
|
}
|
|
|
|
ENDING_SUMMARIES = {
|
|
"escaped": "你们揭穿了旅店真相并组织逃离，雾港旅店在身后沉入迷雾。",
|
|
"sacrifice": "真相被揭开，但有人留下挡住了灾厄，你们带着代价离开。",
|
|
"trapped": "危险失控，旅店的规则吞没了所有出口，你们被永困其中。",
|
|
"pact": "你们选择与隐藏势力交易，离开旅店，却成为新的规则执行者。",
|
|
"team_wipe": "队伍全员倒下，冒险在黑暗中终止。",
|
|
}
|
|
|
|
|
|
class StoryEngine:
|
|
def initial_state(
|
|
self,
|
|
room_id: str,
|
|
player_ids: list[str],
|
|
game_card: dict[str, Any] | None = None,
|
|
) -> GameState:
|
|
inventory = {player_id: [] for player_id in player_ids}
|
|
card = game_card or {}
|
|
opening_text = str(card.get("opening_text") or "暴风雨夜，你们推开雾港旅店大门，空气里有潮湿铁锈味。")
|
|
starter_goals = [str(item) for item in card.get("starter_goals", []) if isinstance(item, str)]
|
|
game_card_id = str(card.get("card_id", "fogharbor"))
|
|
game_card_title = str(card.get("title", "雾港旅店"))
|
|
state_schema = dict(card.get("state_schema", {}))
|
|
content_pack = dict(card.get("content_pack", {}))
|
|
data_scope = self._normalize_data_scope(state_schema.get("data_scope", {}))
|
|
visibility = self._normalize_visibility(state_schema.get("visibility", {}))
|
|
systems_payload = content_pack.get("systems", {}) if isinstance(content_pack, dict) else {}
|
|
runtime_options = self._normalize_runtime_options(
|
|
systems_payload.get("runtime_options", {}) if isinstance(systems_payload, dict) else {}
|
|
)
|
|
card_mode = "fogharbor" if game_card_id == "fogharbor" else "custom"
|
|
initial_scene = "lobby" if card_mode == "fogharbor" else "opening"
|
|
initial_choices = deepcopy(SCENES["lobby"]["choices"]) if card_mode == "fogharbor" else (
|
|
starter_goals[:3] if starter_goals else ["先观察环境", "与关键角色对话", "团队分工行动"]
|
|
)
|
|
|
|
metric_values: dict[str, int] = {}
|
|
for metric in state_schema.get("metrics", []):
|
|
key = str(metric.get("key", "")).strip()
|
|
if not key:
|
|
continue
|
|
metric_values[key] = int(metric.get("initial", 0))
|
|
|
|
metric_values_by_player: dict[str, dict[str, int]] = {}
|
|
if data_scope["metrics"] == "per_player":
|
|
for player_id in player_ids:
|
|
metric_values_by_player[player_id] = dict(metric_values)
|
|
|
|
shared_inventory: list[str] = [] if data_scope["inventory"] == "shared" else []
|
|
clues_by_player: dict[str, list[str]] = {player_id: [] for player_id in player_ids} if data_scope["clues"] == "per_player" else {}
|
|
player_states = {
|
|
player_id: {
|
|
"status": "alive",
|
|
"is_alive": True,
|
|
"deaths": 0,
|
|
"revives": 0,
|
|
"last_event": "spawn",
|
|
"updated_at": utc_now_iso(),
|
|
}
|
|
for player_id in player_ids
|
|
}
|
|
|
|
return GameState(
|
|
room_id=room_id,
|
|
game_phase="lobby",
|
|
chapter=0,
|
|
scene_id=initial_scene,
|
|
scene_description=opening_text,
|
|
global_flags={
|
|
"truth_revealed": False,
|
|
"escape_route_open": False,
|
|
"found_service_key": False,
|
|
"room_301_unlocked": False,
|
|
"cellar_entry_found": False,
|
|
"someone_sacrificed": False,
|
|
"allied_hidden_force": False,
|
|
"game_card_id": game_card_id,
|
|
"game_card_title": game_card_title,
|
|
"starter_goals": starter_goals,
|
|
"card_mode": card_mode,
|
|
"state_schema": state_schema,
|
|
"card_content_pack": content_pack,
|
|
"data_scope": data_scope,
|
|
"visibility": visibility,
|
|
"runtime_options": runtime_options,
|
|
"metric_values": metric_values,
|
|
"metric_values_by_player": metric_values_by_player,
|
|
"shared_inventory": shared_inventory,
|
|
"clues_by_player": clues_by_player,
|
|
"player_states": player_states,
|
|
},
|
|
discovered_clues=[],
|
|
inventory_by_player=inventory,
|
|
npc_states={
|
|
"innkeeper": {"mood": "guarded", "trust": 1, "location": "lobby"},
|
|
"maid": {"mood": "anxious", "trust": 1, "location": "dining_room"},
|
|
"silent_traveler": {"mood": "cold", "trust": 0, "location": "lobby"},
|
|
},
|
|
danger_level=1,
|
|
action_history=[],
|
|
pending_choices=initial_choices,
|
|
ending_state=None,
|
|
round_no=0,
|
|
)
|
|
|
|
def story_context(self, scene_id: str, state: dict[str, Any] | None = None) -> dict[str, Any]:
|
|
scene = SCENES.get(scene_id, SCENES["lobby"])
|
|
global_flags = (state or {}).get("global_flags", {})
|
|
card_mode = global_flags.get("card_mode", "fogharbor")
|
|
runtime_options = self._normalize_runtime_options(global_flags.get("runtime_options", {}))
|
|
if card_mode != "fogharbor":
|
|
content_pack = global_flags.get("card_content_pack", {})
|
|
return {
|
|
"setting": global_flags.get("game_card_title", "自定义剧本"),
|
|
"scene": {
|
|
"id": scene_id,
|
|
"title": scene_id,
|
|
"description": (state or {}).get("scene_description", ""),
|
|
"default_choices": list((state or {}).get("pending_choices", [])),
|
|
},
|
|
"content_pack": content_pack,
|
|
"runtime_options": runtime_options,
|
|
"allowed_clues": "ANY_STRING",
|
|
"npcs": "ANY_NPC",
|
|
"rules": [
|
|
"这是自定义游戏卡，状态字段以 state_schema 为准",
|
|
"玩家/共享数据范围以 state_schema.data_scope 为准",
|
|
"不要强制套用雾港旅店的固定场景和真相",
|
|
"结局判定优先由 content_pack.end_conditions 与服务端规则执行",
|
|
"输出必须仍满足 JSON schema",
|
|
],
|
|
}
|
|
return {
|
|
"setting": "雾港旅店",
|
|
"scene": {
|
|
"id": scene_id,
|
|
"title": scene["title"],
|
|
"description": scene["description"],
|
|
"default_choices": scene["choices"],
|
|
},
|
|
"allowed_clues": sorted(ALLOWED_CLUES),
|
|
"npcs": sorted(NPC_IDS),
|
|
"endings": list(ENDING_SUMMARIES.keys()),
|
|
"runtime_options": runtime_options,
|
|
"rules": [
|
|
"关键真相与结局由服务端状态机控制",
|
|
"危险值达到 10 会触发被困结局",
|
|
"多人冲突由系统先裁定，再交给 AI 叙述",
|
|
],
|
|
}
|
|
|
|
def scene_guidance(self, scene_id: str, round_no: int, pending_choices: list[str]) -> list[str]:
|
|
if round_no == 0:
|
|
return [
|
|
"先阅读场景描述，确认当前最可疑的人或地点。",
|
|
"至少一名玩家提交“调查/交谈”类行动，触发第一轮反馈。",
|
|
"多人局建议先分工：一人问 NPC，一人搜线索，一人记录关键信息。",
|
|
]
|
|
if pending_choices:
|
|
return [f"可优先尝试：{choice}" for choice in pending_choices[:3]]
|
|
scene = SCENES.get(scene_id, SCENES["lobby"])
|
|
return [f"继续推进：{choice}" for choice in scene["choices"][:3]]
|
|
|
|
def apply_turn(
|
|
self,
|
|
state_dict: dict[str, Any],
|
|
ai_payload: dict[str, Any],
|
|
merge_result: dict[str, Any],
|
|
players: list[dict[str, Any]],
|
|
) -> tuple[dict[str, Any], dict[str, Any]]:
|
|
state = GameState.from_dict(state_dict)
|
|
player_ids = {player["player_id"] for player in players}
|
|
|
|
for player in players:
|
|
state.inventory_by_player.setdefault(player["player_id"], [])
|
|
|
|
updates = ai_payload.get("state_updates", {})
|
|
card_mode = state.global_flags.get("card_mode", "fogharbor")
|
|
data_scope = self._normalize_data_scope(state.global_flags.get("data_scope", {}))
|
|
runtime_options = self._normalize_runtime_options(state.global_flags.get("runtime_options", {}))
|
|
state.global_flags["runtime_options"] = runtime_options
|
|
player_ids_list = [player["player_id"] for player in players]
|
|
danger_delta = int(updates.get("danger_delta", 0))
|
|
if card_mode == "fogharbor":
|
|
state.danger_level = max(0, min(10, state.danger_level + danger_delta))
|
|
|
|
clue_targets = merge_result.get("triggered_players", [])
|
|
if not isinstance(clue_targets, list) or not clue_targets:
|
|
clue_targets = player_ids_list
|
|
for clue in updates.get("add_clues", []):
|
|
if not isinstance(clue, str):
|
|
continue
|
|
clue_value = clue.strip()
|
|
if not clue_value:
|
|
continue
|
|
if card_mode == "fogharbor" and clue_value not in ALLOWED_CLUES:
|
|
continue
|
|
if data_scope["clues"] == "per_player":
|
|
clues_by_player = state.global_flags.get("clues_by_player", {})
|
|
if not isinstance(clues_by_player, dict):
|
|
clues_by_player = {}
|
|
for target_id in clue_targets:
|
|
if target_id not in player_ids:
|
|
continue
|
|
rows = clues_by_player.get(target_id, [])
|
|
if clue_value not in rows:
|
|
rows.append(clue_value)
|
|
clues_by_player[target_id] = rows
|
|
state.global_flags["clues_by_player"] = clues_by_player
|
|
else:
|
|
if clue_value not in state.discovered_clues:
|
|
state.discovered_clues.append(clue_value)
|
|
|
|
flags = updates.get("set_flags", {})
|
|
for key, value in flags.items():
|
|
if isinstance(key, str):
|
|
state.global_flags[key] = value
|
|
|
|
npc_changes = updates.get("npc_changes", {})
|
|
for npc_id, patch in npc_changes.items():
|
|
if npc_id not in NPC_IDS or not isinstance(patch, dict):
|
|
continue
|
|
state.npc_states.setdefault(npc_id, {})
|
|
for patch_key, patch_value in patch.items():
|
|
state.npc_states[npc_id][patch_key] = patch_value
|
|
|
|
inventory_add = updates.get("inventory_add", {})
|
|
inventory_remove = updates.get("inventory_remove", {})
|
|
if data_scope["inventory"] == "shared":
|
|
shared_inventory = state.global_flags.get("shared_inventory", [])
|
|
if not isinstance(shared_inventory, list):
|
|
shared_inventory = []
|
|
for _, items in inventory_add.items():
|
|
if not isinstance(items, list):
|
|
continue
|
|
for item in items:
|
|
if not isinstance(item, str):
|
|
continue
|
|
item_value = self._normalize_inventory_item_key(item.strip(), card_mode)
|
|
if not item_value:
|
|
continue
|
|
if card_mode == "fogharbor" and item_value not in ALLOWED_CLUES:
|
|
continue
|
|
if item_value not in shared_inventory:
|
|
shared_inventory.append(item_value)
|
|
for _, items in inventory_remove.items():
|
|
if not isinstance(items, list):
|
|
continue
|
|
for item in items:
|
|
if not isinstance(item, str):
|
|
continue
|
|
item_value = self._normalize_inventory_item_key(item.strip(), card_mode)
|
|
if not item_value:
|
|
continue
|
|
if card_mode == "fogharbor" and item_value not in ALLOWED_CLUES:
|
|
continue
|
|
shared_inventory = [row for row in shared_inventory if row != item_value]
|
|
state.global_flags["shared_inventory"] = shared_inventory
|
|
else:
|
|
for player_id, items in inventory_add.items():
|
|
if player_id not in player_ids:
|
|
continue
|
|
state.inventory_by_player.setdefault(player_id, [])
|
|
for item in items:
|
|
if not isinstance(item, str):
|
|
continue
|
|
item_value = self._normalize_inventory_item_key(item.strip(), card_mode)
|
|
if not item_value:
|
|
continue
|
|
if card_mode == "fogharbor" and item_value not in ALLOWED_CLUES:
|
|
continue
|
|
if item_value not in state.inventory_by_player[player_id]:
|
|
state.inventory_by_player[player_id].append(item_value)
|
|
for player_id, items in inventory_remove.items():
|
|
target_players = [player_id]
|
|
if player_id.strip().lower() in {"*", "all"}:
|
|
target_players = list(player_ids)
|
|
for target_player_id in target_players:
|
|
if target_player_id not in player_ids:
|
|
continue
|
|
state.inventory_by_player.setdefault(target_player_id, [])
|
|
if not isinstance(items, list):
|
|
continue
|
|
for item in items:
|
|
if not isinstance(item, str):
|
|
continue
|
|
item_value = self._normalize_inventory_item_key(item.strip(), card_mode)
|
|
if not item_value:
|
|
continue
|
|
if card_mode == "fogharbor" and item_value not in ALLOWED_CLUES:
|
|
continue
|
|
state.inventory_by_player[target_player_id] = [
|
|
row for row in state.inventory_by_player[target_player_id] if row != item_value
|
|
]
|
|
if card_mode != "fogharbor":
|
|
self._apply_custom_metrics(state, updates, flags, players, merge_result)
|
|
|
|
self._apply_player_survival_updates(state, flags, players, runtime_options)
|
|
|
|
state.scene_description = ai_payload.get("narration", state.scene_description)
|
|
state.pending_choices = ai_payload.get("choices", state.pending_choices)[:4]
|
|
state.round_no += 1
|
|
state.last_updated_at = utc_now_iso()
|
|
|
|
history_row = {
|
|
"round": state.round_no,
|
|
"scene_id": state.scene_id,
|
|
"merged_actions": merge_result,
|
|
"narration": ai_payload.get("narration", ""),
|
|
"npc_dialogues": ai_payload.get("npc_dialogues", []),
|
|
}
|
|
state.action_history.append(history_row)
|
|
if len(state.action_history) > 50:
|
|
state.action_history = state.action_history[-50:]
|
|
|
|
transitioned = None
|
|
ending_state = None
|
|
if card_mode == "fogharbor":
|
|
transitioned = self._auto_progress_scene(state)
|
|
ending_state = self._check_endings(state)
|
|
else:
|
|
transitioned = self._custom_scene_progress(state, ai_payload)
|
|
ending_state = self._check_custom_endings(state, ai_payload)
|
|
if (
|
|
runtime_options.get("allow_player_death")
|
|
and not ending_state
|
|
and self._alive_player_count(state) <= 0
|
|
):
|
|
ending_state = "team_wipe"
|
|
if not str(state.global_flags.get("ending_summary", "")).strip():
|
|
state.global_flags["ending_summary"] = ENDING_SUMMARIES["team_wipe"]
|
|
if ending_state:
|
|
state.game_phase = "ended"
|
|
state.ending_state = ending_state
|
|
ending_summary = ai_payload.get("ending_summary")
|
|
if isinstance(ending_summary, str) and ending_summary.strip():
|
|
state.global_flags["ending_summary"] = ending_summary.strip()
|
|
elif not str(state.global_flags.get("ending_summary", "")).strip():
|
|
fallback_summary = ENDING_SUMMARIES.get(ending_state, "")
|
|
if fallback_summary:
|
|
state.global_flags["ending_summary"] = fallback_summary
|
|
|
|
round_result = {
|
|
"narration": ai_payload.get("narration", ""),
|
|
"npc_dialogues": ai_payload.get("npc_dialogues", []),
|
|
"private_player_updates": ai_payload.get("private_player_updates", []),
|
|
"choices": state.pending_choices,
|
|
"merge_summary": merge_result.get("summary", ""),
|
|
"conflicts": merge_result.get("conflicts", []),
|
|
"triggered_players": merge_result.get("triggered_players", []),
|
|
"scene_transition": transitioned,
|
|
"ending_state": state.ending_state,
|
|
"ending_summary": str(state.global_flags.get("ending_summary", "")).strip() or ENDING_SUMMARIES.get(state.ending_state),
|
|
"player_states": state.global_flags.get("player_states", {}),
|
|
"runtime_options": runtime_options,
|
|
"guidance": self.scene_guidance(state.scene_id, state.round_no, state.pending_choices),
|
|
}
|
|
return state.to_dict(), round_result
|
|
|
|
def _apply_custom_metrics(
|
|
self,
|
|
state: GameState,
|
|
updates: dict[str, Any],
|
|
flags: dict[str, Any],
|
|
players: list[dict[str, Any]],
|
|
merge_result: dict[str, Any],
|
|
) -> None:
|
|
schema = dict(state.global_flags.get("state_schema", {}))
|
|
metrics = schema.get("metrics", [])
|
|
data_scope = self._normalize_data_scope(state.global_flags.get("data_scope", {}))
|
|
player_ids = [player["player_id"] for player in players]
|
|
triggered_players = merge_result.get("triggered_players", [])
|
|
if not isinstance(triggered_players, list) or not triggered_players:
|
|
triggered_players = list(player_ids)
|
|
|
|
if data_scope["metrics"] == "per_player":
|
|
self._apply_player_metrics(state, metrics, flags, updates, player_ids, triggered_players)
|
|
return
|
|
|
|
metric_values = dict(state.global_flags.get("metric_values", {}))
|
|
for metric in metrics:
|
|
key = str(metric.get("key", "")).strip()
|
|
if not key:
|
|
continue
|
|
current_value = int(metric_values.get(key, 0))
|
|
if isinstance(flags.get(key), (int, float)):
|
|
current_value = int(flags[key])
|
|
delta_key = f"{key}_delta"
|
|
if isinstance(flags.get(delta_key), (int, float)):
|
|
current_value += int(flags[delta_key])
|
|
|
|
if key == "danger_level":
|
|
current_value += int(updates.get("danger_delta", 0))
|
|
|
|
max_value = int(metric.get("max", 10))
|
|
min_value = int(metric.get("min", 0))
|
|
current_value = max(min_value, min(max_value, current_value))
|
|
metric_values[key] = current_value
|
|
|
|
state.global_flags["metric_values"] = metric_values
|
|
|
|
def _apply_player_metrics(
|
|
self,
|
|
state: GameState,
|
|
metrics: list[dict[str, Any]],
|
|
flags: dict[str, Any],
|
|
updates: dict[str, Any],
|
|
player_ids: list[str],
|
|
triggered_players: list[str],
|
|
) -> None:
|
|
map_by_player = state.global_flags.get("metric_values_by_player", {})
|
|
if not isinstance(map_by_player, dict):
|
|
map_by_player = {}
|
|
default_metric_values = state.global_flags.get("metric_values", {})
|
|
if not isinstance(default_metric_values, dict):
|
|
default_metric_values = {}
|
|
|
|
for player_id in player_ids:
|
|
player_metrics = map_by_player.get(player_id)
|
|
if not isinstance(player_metrics, dict):
|
|
player_metrics = dict(default_metric_values)
|
|
for metric in metrics:
|
|
key = str(metric.get("key", "")).strip()
|
|
if not key:
|
|
continue
|
|
if key not in player_metrics:
|
|
player_metrics[key] = int(metric.get("initial", 0))
|
|
map_by_player[player_id] = player_metrics
|
|
|
|
player_metric_set = flags.get("player_metric_set", {})
|
|
if not isinstance(player_metric_set, dict):
|
|
player_metric_set = {}
|
|
player_metric_delta = flags.get("player_metric_delta", {})
|
|
if not isinstance(player_metric_delta, dict):
|
|
player_metric_delta = {}
|
|
|
|
for metric in metrics:
|
|
key = str(metric.get("key", "")).strip()
|
|
if not key:
|
|
continue
|
|
min_value = int(metric.get("min", 0))
|
|
max_value = int(metric.get("max", 10))
|
|
if max_value < min_value:
|
|
max_value = min_value
|
|
|
|
for player_id in player_ids:
|
|
value = int(map_by_player[player_id].get(key, 0))
|
|
|
|
set_payload = player_metric_set.get(player_id, {})
|
|
if isinstance(set_payload, dict) and isinstance(set_payload.get(key), (int, float)):
|
|
value = int(set_payload[key])
|
|
|
|
delta_payload = player_metric_delta.get(player_id, {})
|
|
if isinstance(delta_payload, dict) and isinstance(delta_payload.get(key), (int, float)):
|
|
value += int(delta_payload[key])
|
|
|
|
if player_id in triggered_players:
|
|
if isinstance(flags.get(key), (int, float)):
|
|
value = int(flags[key])
|
|
delta_key = f"{key}_delta"
|
|
if isinstance(flags.get(delta_key), (int, float)):
|
|
value += int(flags[delta_key])
|
|
if key == "danger_level":
|
|
value += int(updates.get("danger_delta", 0))
|
|
|
|
value = max(min_value, min(max_value, value))
|
|
map_by_player[player_id][key] = value
|
|
|
|
state.global_flags["metric_values_by_player"] = map_by_player
|
|
|
|
def _alive_player_count(self, state: GameState) -> int:
|
|
player_states = state.global_flags.get("player_states", {})
|
|
if not isinstance(player_states, dict):
|
|
return 0
|
|
alive = 0
|
|
for row in player_states.values():
|
|
if not isinstance(row, dict):
|
|
continue
|
|
if bool(row.get("is_alive", True)):
|
|
alive += 1
|
|
return alive
|
|
|
|
def _safe_int(self, value: Any, default: int = 0) -> int:
|
|
try:
|
|
return int(value)
|
|
except (TypeError, ValueError):
|
|
return int(default)
|
|
|
|
def _normalize_runtime_options(self, raw_options: Any) -> dict[str, Any]:
|
|
options = raw_options if isinstance(raw_options, dict) else {}
|
|
|
|
def _to_bool(value: Any, default: bool) -> bool:
|
|
if isinstance(value, bool):
|
|
return value
|
|
text = str(value or "").strip().lower()
|
|
if text in {"1", "true", "yes", "on", "y"}:
|
|
return True
|
|
if text in {"0", "false", "no", "off", "n"}:
|
|
return False
|
|
return default
|
|
|
|
interaction_mode = str(options.get("interaction_mode", "hybrid")).strip().lower()
|
|
if interaction_mode not in {"cooperative", "split", "hybrid"}:
|
|
interaction_mode = "hybrid"
|
|
return {
|
|
"interaction_mode": interaction_mode,
|
|
"sync_all_player_ai_replies": _to_bool(options.get("sync_all_player_ai_replies", False), False),
|
|
"allow_player_death": _to_bool(options.get("allow_player_death", False), False),
|
|
"allow_revive": _to_bool(options.get("allow_revive", False), False),
|
|
"require_all_players_response": _to_bool(options.get("require_all_players_response", False), False),
|
|
"response_timeout_seconds": max(5, min(300, self._safe_int(options.get("response_timeout_seconds", 45), 45))),
|
|
}
|
|
|
|
def _ensure_player_states(self, state: GameState, players: list[dict[str, Any]]) -> dict[str, dict[str, Any]]:
|
|
raw_states = state.global_flags.get("player_states", {})
|
|
if not isinstance(raw_states, dict):
|
|
raw_states = {}
|
|
now = utc_now_iso()
|
|
out: dict[str, dict[str, Any]] = {}
|
|
for player in players:
|
|
player_id = str(player.get("player_id", "")).strip()
|
|
if not player_id:
|
|
continue
|
|
row = raw_states.get(player_id, {})
|
|
if not isinstance(row, dict):
|
|
row = {}
|
|
status = str(row.get("status", "")).strip().lower()
|
|
if status not in {"alive", "down", "dead", "failed"}:
|
|
status = "alive"
|
|
is_alive_raw = row.get("is_alive")
|
|
if isinstance(is_alive_raw, bool):
|
|
is_alive = is_alive_raw
|
|
else:
|
|
is_alive = status not in {"dead", "failed"}
|
|
out[player_id] = {
|
|
"status": status,
|
|
"is_alive": is_alive,
|
|
"deaths": max(0, self._safe_int(row.get("deaths", 0), 0)),
|
|
"revives": max(0, self._safe_int(row.get("revives", 0), 0)),
|
|
"last_event": str(row.get("last_event", "")).strip()[:24],
|
|
"updated_at": str(row.get("updated_at", "")).strip() or now,
|
|
}
|
|
state.global_flags["player_states"] = out
|
|
return out
|
|
|
|
def _iter_flagged_players(self, value: Any) -> list[str]:
|
|
out: list[str] = []
|
|
if isinstance(value, dict):
|
|
for key, enabled in value.items():
|
|
if isinstance(enabled, bool):
|
|
if enabled:
|
|
out.append(str(key).strip())
|
|
elif isinstance(enabled, (int, float)):
|
|
if int(enabled) != 0:
|
|
out.append(str(key).strip())
|
|
elif str(enabled or "").strip().lower() in {"1", "true", "yes", "on", "y"}:
|
|
out.append(str(key).strip())
|
|
elif isinstance(value, list):
|
|
for row in value:
|
|
player_id = str(row or "").strip()
|
|
if player_id:
|
|
out.append(player_id)
|
|
elif isinstance(value, str):
|
|
for row in value.split(","):
|
|
player_id = row.strip()
|
|
if player_id:
|
|
out.append(player_id)
|
|
return out
|
|
|
|
def _apply_player_survival_updates(
|
|
self,
|
|
state: GameState,
|
|
flags: dict[str, Any],
|
|
players: list[dict[str, Any]],
|
|
runtime_options: dict[str, Any],
|
|
) -> None:
|
|
player_states = self._ensure_player_states(state, players)
|
|
allow_death = bool(runtime_options.get("allow_player_death"))
|
|
allow_revive = bool(runtime_options.get("allow_revive"))
|
|
|
|
if not allow_death:
|
|
now = utc_now_iso()
|
|
for row in player_states.values():
|
|
row["status"] = "alive"
|
|
row["is_alive"] = True
|
|
row["updated_at"] = now
|
|
state.global_flags["player_states"] = player_states
|
|
return
|
|
|
|
def mark_dead(player_id: str, *, event: str = "dead") -> None:
|
|
player = player_states.get(player_id)
|
|
if not isinstance(player, dict):
|
|
return
|
|
if bool(player.get("is_alive", True)):
|
|
player["deaths"] = max(0, self._safe_int(player.get("deaths", 0), 0)) + 1
|
|
player["status"] = "dead"
|
|
player["is_alive"] = False
|
|
player["last_event"] = event[:24]
|
|
player["updated_at"] = utc_now_iso()
|
|
|
|
def mark_alive(player_id: str, *, event: str = "revived") -> None:
|
|
player = player_states.get(player_id)
|
|
if not isinstance(player, dict):
|
|
return
|
|
if not bool(player.get("is_alive", True)):
|
|
player["revives"] = max(0, self._safe_int(player.get("revives", 0), 0)) + 1
|
|
player["status"] = "alive"
|
|
player["is_alive"] = True
|
|
player["last_event"] = event[:24]
|
|
player["updated_at"] = utc_now_iso()
|
|
|
|
status_patch = flags.get("player_status", {})
|
|
if isinstance(status_patch, dict):
|
|
for raw_player_id, raw_status in status_patch.items():
|
|
player_id = str(raw_player_id or "").strip()
|
|
status = str(raw_status or "").strip().lower()
|
|
if player_id not in player_states or not status:
|
|
continue
|
|
if status in {"dead", "failed"}:
|
|
mark_dead(player_id, event=status)
|
|
elif status in {"alive", "revived"} and allow_revive:
|
|
mark_alive(player_id, event=status)
|
|
elif status == "down":
|
|
player = player_states[player_id]
|
|
player["status"] = "down"
|
|
player["is_alive"] = False
|
|
player["last_event"] = "down"
|
|
player["updated_at"] = utc_now_iso()
|
|
|
|
for player_id in self._iter_flagged_players(flags.get("player_dead", {})):
|
|
if player_id in player_states:
|
|
mark_dead(player_id, event="dead")
|
|
for player_id in self._iter_flagged_players(flags.get("player_death", {})):
|
|
if player_id in player_states:
|
|
mark_dead(player_id, event="dead")
|
|
|
|
survival_patch = flags.get("player_survival", {})
|
|
if isinstance(survival_patch, dict):
|
|
for raw_player_id, raw_alive in survival_patch.items():
|
|
player_id = str(raw_player_id or "").strip()
|
|
if player_id not in player_states:
|
|
continue
|
|
alive = None
|
|
if isinstance(raw_alive, bool):
|
|
alive = raw_alive
|
|
elif isinstance(raw_alive, (int, float)):
|
|
alive = int(raw_alive) != 0
|
|
else:
|
|
text = str(raw_alive or "").strip().lower()
|
|
if text in {"1", "true", "yes", "on", "y"}:
|
|
alive = True
|
|
elif text in {"0", "false", "no", "off", "n"}:
|
|
alive = False
|
|
if alive is None:
|
|
continue
|
|
if alive:
|
|
if allow_revive:
|
|
mark_alive(player_id, event="revived")
|
|
else:
|
|
mark_dead(player_id, event="failed")
|
|
|
|
if allow_revive:
|
|
for player_id in self._iter_flagged_players(flags.get("player_revive", {})):
|
|
if player_id in player_states:
|
|
mark_alive(player_id, event="revived")
|
|
|
|
state.global_flags["player_states"] = player_states
|
|
|
|
def _normalize_data_scope(self, raw_scope: Any) -> dict[str, str]:
|
|
scope = raw_scope if isinstance(raw_scope, dict) else {}
|
|
out = {
|
|
"metrics": str(scope.get("metrics", "shared")).strip().lower(),
|
|
"inventory": str(scope.get("inventory", "per_player")).strip().lower(),
|
|
"clues": str(scope.get("clues", "shared")).strip().lower(),
|
|
}
|
|
for key in ("metrics", "inventory", "clues"):
|
|
if out[key] not in {"shared", "per_player"}:
|
|
out[key] = "shared" if key != "inventory" else "per_player"
|
|
return out
|
|
|
|
def _normalize_visibility(self, raw_visibility: Any) -> dict[str, bool]:
|
|
visibility = raw_visibility if isinstance(raw_visibility, dict) else {}
|
|
def _to_bool(value: Any, default: bool) -> bool:
|
|
if isinstance(value, bool):
|
|
return value
|
|
text = str(value or "").strip().lower()
|
|
if text in {"1", "true", "yes", "on", "y"}:
|
|
return True
|
|
if text in {"0", "false", "no", "off", "n"}:
|
|
return False
|
|
return default
|
|
return {
|
|
"show_all_player_metrics": _to_bool(visibility.get("show_all_player_metrics", True), True),
|
|
"show_all_player_inventory": _to_bool(visibility.get("show_all_player_inventory", True), True),
|
|
}
|
|
|
|
def _normalize_inventory_item_key(self, item_value: str, card_mode: str) -> str:
|
|
value = str(item_value or "").strip()
|
|
if not value:
|
|
return ""
|
|
if card_mode == "fogharbor":
|
|
return FOGHARBOR_ITEM_ALIASES.get(value, value)
|
|
return value
|
|
|
|
def _custom_scene_progress(self, state: GameState, ai_payload: dict[str, Any]) -> dict[str, Any] | None:
|
|
if not ai_payload.get("should_end_scene"):
|
|
return None
|
|
prev_scene = state.scene_id
|
|
state.chapter = int(state.chapter) + 1
|
|
state.scene_id = f"scene_{state.chapter}"
|
|
return {
|
|
"from": prev_scene,
|
|
"to": state.scene_id,
|
|
"chapter": state.chapter,
|
|
"description": state.scene_description,
|
|
}
|
|
|
|
def _check_custom_endings(self, state: GameState, ai_payload: dict[str, Any]) -> str | None:
|
|
configured = self._evaluate_custom_end_conditions(state)
|
|
if configured:
|
|
ending_state, ending_summary = configured
|
|
state.global_flags["custom_ending_state"] = ending_state
|
|
if ending_summary:
|
|
state.global_flags["ending_summary"] = ending_summary
|
|
return ending_state
|
|
|
|
if ai_payload.get("should_end_game"):
|
|
custom = state.global_flags.get("custom_ending_state")
|
|
if isinstance(custom, str) and custom.strip():
|
|
return custom.strip()
|
|
return "custom_end"
|
|
return None
|
|
|
|
def _evaluate_custom_end_conditions(self, state: GameState) -> tuple[str, str] | None:
|
|
content_pack = state.global_flags.get("card_content_pack", {})
|
|
if not isinstance(content_pack, dict):
|
|
return None
|
|
conditions = content_pack.get("end_conditions", [])
|
|
if not isinstance(conditions, list):
|
|
return None
|
|
|
|
metrics = state.global_flags.get("metric_values", {})
|
|
if not isinstance(metrics, dict):
|
|
metrics = {}
|
|
|
|
for row in conditions[:40]:
|
|
if not isinstance(row, dict):
|
|
continue
|
|
|
|
has_clause = False
|
|
passed = True
|
|
|
|
round_gate = row.get("round")
|
|
if isinstance(round_gate, (int, float)):
|
|
has_clause = True
|
|
if int(state.round_no) < int(round_gate):
|
|
passed = False
|
|
|
|
metric_key = str(row.get("metric", "")).strip()
|
|
if metric_key:
|
|
has_clause = True
|
|
metric_value = metrics.get(metric_key, state.global_flags.get(metric_key))
|
|
expected_value = row.get("value")
|
|
op = str(row.get("op", ">=")).strip() or ">="
|
|
if not self._compare_condition(metric_value, op, expected_value):
|
|
passed = False
|
|
|
|
flag_key = str(row.get("flag", "")).strip()
|
|
if flag_key:
|
|
has_clause = True
|
|
expected_flag = row.get("flag_equals", True)
|
|
actual_flag = state.global_flags.get(flag_key)
|
|
if actual_flag != expected_flag:
|
|
passed = False
|
|
|
|
if not has_clause or not passed:
|
|
continue
|
|
|
|
ending_state = str(row.get("ending_state", "")).strip() or str(row.get("id", "")).strip() or "custom_end"
|
|
ending_summary = str(row.get("ending_summary", "")).strip()
|
|
return ending_state, ending_summary
|
|
return None
|
|
|
|
def _compare_condition(self, actual: Any, op: str, expected: Any) -> bool:
|
|
numeric_ops = {">", ">=", "<", "<="}
|
|
if op in numeric_ops:
|
|
if not isinstance(actual, (int, float)) or not isinstance(expected, (int, float)):
|
|
return False
|
|
if op == ">":
|
|
return actual > expected
|
|
if op == ">=":
|
|
return actual >= expected
|
|
if op == "<":
|
|
return actual < expected
|
|
if op == "<=":
|
|
return actual <= expected
|
|
|
|
if op == "==":
|
|
return actual == expected
|
|
if op == "!=":
|
|
return actual != expected
|
|
return False
|
|
|
|
def _auto_progress_scene(self, state: GameState) -> dict[str, Any] | None:
|
|
prev_scene = state.scene_id
|
|
flags = state.global_flags
|
|
clues = set(state.discovered_clues)
|
|
|
|
if state.scene_id == "lobby" and (
|
|
"torn_manifest" in clues or "missing_guest_note" in clues or flags.get("found_service_key")
|
|
):
|
|
state.scene_id = "dining_room"
|
|
elif state.scene_id == "dining_room" and (
|
|
flags.get("found_service_key") or "silver_bell" in clues or flags.get("room_301_unlocked")
|
|
):
|
|
state.scene_id = "corridor"
|
|
elif state.scene_id == "corridor" and (
|
|
flags.get("room_301_unlocked") or "maid_diary" in clues
|
|
):
|
|
state.scene_id = "guest_room"
|
|
elif state.scene_id == "guest_room" and (
|
|
flags.get("cellar_entry_found") or "cellar_sigil" in clues
|
|
):
|
|
state.scene_id = "basement"
|
|
|
|
if state.scene_id != prev_scene:
|
|
scene = SCENES[state.scene_id]
|
|
state.chapter = int(scene["chapter"])
|
|
state.scene_description = scene["description"]
|
|
if not state.pending_choices:
|
|
state.pending_choices = deepcopy(scene["choices"])
|
|
return {
|
|
"from": prev_scene,
|
|
"to": state.scene_id,
|
|
"chapter": state.chapter,
|
|
"description": scene["description"],
|
|
}
|
|
|
|
if state.chapter == 0:
|
|
state.chapter = 1
|
|
return None
|
|
|
|
def _check_endings(self, state: GameState) -> str | None:
|
|
flags = state.global_flags
|
|
clues = set(state.discovered_clues)
|
|
|
|
if state.danger_level >= 10:
|
|
return "trapped"
|
|
|
|
if (
|
|
state.scene_id == "basement"
|
|
and flags.get("truth_revealed")
|
|
and flags.get("escape_route_open")
|
|
and state.danger_level <= 7
|
|
and len(clues) >= 4
|
|
):
|
|
return "escaped"
|
|
|
|
if flags.get("truth_revealed") and flags.get("someone_sacrificed"):
|
|
return "sacrifice"
|
|
|
|
if state.scene_id == "basement" and flags.get("allied_hidden_force"):
|
|
return "pact"
|
|
|
|
return None
|