"""回合行动输入整理与前置处理。""" from __future__ import annotations import logging import re from typing import Any from src.engine import progression from src.engine.checks import build_check_request, roll_check_request from src.engine.character_utils import is_conscious from src.engine.game_instance import GameInstance from src.engine import module_hooks from src.engine.language import localized_text from src.content_modules.projection import entries_for_world logger = logging.getLogger("trpg") def ensure_round_managers(instance: GameInstance) -> None: """确保回合处理需要的运行期管理器已存在。""" instance.ensure_round_managers() def format_action_line(instance: GameInstance, action: dict) -> str: """把单个行动格式化成给 GM/LLM 看的归因文本。""" name = instance.players[action['user_id']].get('character_name', action['user_id']) text = action.get('text', '') parts = [] selected_attr = action.get('selected_attribute', '') selected_skill = action.get('selected_skill', '') if selected_attr or selected_skill: label = localized_text(instance.language, {"en": "Check", "zh-CN": "检定", "ja": "判定", "de": "Probe"}) parts.append(f"{label}:{selected_attr or '?'}" + (f"/{selected_skill}" if selected_skill else "")) target = action.get('target_text', '') if target: label = localized_text(instance.language, {"en": "Target", "zh-CN": "目标", "ja": "対象", "de": "Ziel"}) parts.append(f"{label}:{target}") tag = f" [{' '.join(parts)}]" if parts else "" return f"【{name}】{text}{tag}" def collect_actions_text(instance: GameInstance) -> str: """收集本轮玩家行动；GM 私密指令不得参与行动/检定识别。""" player_lines = [ format_action_line(instance, action) for action in instance.action_queue if action.get("user_id") in instance.players and is_conscious(instance.get_character_sheet(str(action.get("user_id") or ""))) ] actions_text = "\n".join(player_lines) if not actions_text: actions_text = "本轮没有玩家行动。" return actions_text def collect_gm_directives_text(instance: GameInstance) -> tuple[str, list[str]]: """返回本轮可用的 GM 私密指令文本及其 ID，供成功结算后消费。""" current_round = progression.current_round(instance) entries = [ entry for entry in module_hooks.read(instance, module_hooks.GM_DIRECTIVES) if int(entry.get("target_round", current_round) or current_round) <= current_round ] if not entries: return "", [] heading = localized_text(instance.language, { "en": "[Private GM Directives]", "zh-CN": "【GM私密指令】", "ja": "【GMプライベート指示】", "de": "[Private GM-Anweisungen]", }) requirement = localized_text(instance.language, { "en": ( "Apply these directives to narration only. Never quote or reveal them to players. " "They must not change, override, or regenerate any server-provided CheckResult." ), "zh-CN": "以下内容只用于修正本轮叙事，禁止向玩家复述、引用或展示；" "不得改变、覆盖或重新生成任何由服务端给出的 CheckResult。", "ja": ( "これらの指示はナレーションの修正にのみ使用し、プレイヤーに復唱・引用・" "開示してはならない。サーバーが提供した CheckResult を変更・上書き・" "再生成してはならない。" ), "de": ( "Verwende diese Anweisungen nur zur Anpassung der Erzählung. Zitiere oder enthülle sie " "niemals gegenüber den Spielern. Sie dürfen kein von der Server-Seite geliefertes " "CheckResult ändern, überschreiben oder neu erzeugen." ), }) body = "\n".join(f"- {entry.get('text', '')}" for entry in entries if entry.get("text")) return f"\n\n{heading}\n{requirement}\n{body}", [str(entry.get("id") or "") for entry in entries] def format_check_results_constraint(instance: GameInstance, checks: list[dict]) -> str: """把已结算 CheckResult 重新格式化为 GM 私有硬约束，供 swipe 等重生成复用。""" if not checks: return "" blocks: list[str] = [] for check in checks: dice = str(check.get("dice") or "") roll_value = check.get("roll") if dice == "d100": math_text = f"d100={roll_value} vs {check.get('threshold')}" difficulty = str(check.get("difficulty") or "") if difficulty in {"hard", "extreme"}: math_text += localized_text(instance.language, { "en": f" (needs a {difficulty} success: ≤{check.get('required_threshold')})", "zh-CN": f"（需要{'困难' if difficulty == 'hard' else '极难'}成功：≤{check.get('required_threshold')}）", "ja": f"（{'ハード' if difficulty == 'hard' else 'イクストリーム'}成功が必要：≤{check.get('required_threshold')}）", }) opposed = check.get("opposed") if isinstance(check.get("opposed"), dict) else None if opposed: math_text += localized_text(instance.language, { "en": (f"; opposed by {check.get('opponent_name')} {opposed.get('opponent_skill')} " f"d100={opposed.get('opponent_roll')} vs {opposed.get('opponent_threshold')} " f"({opposed.get('opponent_verdict')})"), "zh-CN": (f"；对抗 {check.get('opponent_name')} 的{opposed.get('opponent_skill')} " f"d100={opposed.get('opponent_roll')} vs {opposed.get('opponent_threshold')}" f"（{opposed.get('opponent_verdict')}）"), "ja": (f"；対抗 {check.get('opponent_name')} の{opposed.get('opponent_skill')} " f"d100={opposed.get('opponent_roll')} vs {opposed.get('opponent_threshold')}" f"（{opposed.get('opponent_verdict')}）"), }) else: modifier = int(check.get("modifier", 0) or 0) total = check.get("total") dc = check.get("dc") if check.get("opponent_name"): math_text = ( f"d20={roll_value} {modifier:+d} = {total} vs " f"{check.get('opponent_name')} d20={check.get('opponent_roll')} " f"{int(check.get('opponent_modifier', 0) or 0):+d} = {check.get('opponent_total')}" ) else: math_text = f"d20={roll_value} {modifier:+d} = {total} vs DC {dc}" blocks.append(localized_text(instance.language, { "en": ( "[System Check - Must Follow]\n" f"Actor: {check.get('actor_name') or check.get('actor_uid')}\n" f"Check: {check.get('label') or dice}: {math_text}\n" f"Result: {check.get('verdict')}\n" "Requirement: keep this server-resolved outcome unchanged." ), "zh-CN": ( "【系统检定·必须遵循】\n" f"角色: {check.get('actor_name') or check.get('actor_uid')}\n" f"检定: {check.get('label') or dice}：{math_text}\n" f"结果: {check.get('verdict')}\n" "要求: 这是服务端已结算结果，不得重掷或改判。" ), "ja": ( "【システム判定・厳守】\n" f"キャラクター: {check.get('actor_name') or check.get('actor_uid')}\n" f"判定: {check.get('label') or dice}：{math_text}\n" f"結果: {check.get('verdict')}\n" "要件: これはサーバーで確定済みの結果であり、再ロールや改変をしてはならない。" ), "de": ( "[Systemseitige Probe - Muss befolgt werden]\n" f"Akteur: {check.get('actor_name') or check.get('actor_uid')}\n" f"Probe: {check.get('label') or dice}: {math_text}\n" f"Ergebnis: {check.get('verdict')}\n" "Anforderung: Dieses vom Server ermittelte Ergebnis bleibt unverändert." ), })) push = check.get("push") if isinstance(push, dict) and push.get("status") == "resolved": import json # An explicit domain record, not executable narrative instructions. blocks.append("[Accepted Push decision] " + json.dumps({ "check_id": check.get("check_id"), "original_roll": push.get("original", {}).get("roll"), "new_roll": check.get("roll"), "new_approach": push.get("reason"), "failed": push.get("failed"), "failure_consequence": push.get("consequence"), }, ensure_ascii=False) + "\nApply the accepted consequence only if Push failed. Do not re-roll or spend Luck. No automatic damage multiplier.") return "\n\n" + "\n\n".join(blocks) def initialize_puzzles_from_lorebook( instance: GameInstance, lorebook_store: Any, *, world_data: dict | None = None, content_projection: Any | None = None, ) -> None: """从世界书初始化谜题（仅新增未注册的谜题）。""" if not instance.world_id or not lorebook_store or not instance.puzzle_manager: return from src.content.worlds import localize_lorebook_entries if content_projection is not None: entries = content_projection.for_game( instance, viewer_kind="gm", ) else: entries = entries_for_world(lorebook_store, instance.world_id) all_entries = localize_lorebook_entries(entries, world_data) for entry in all_entries: if entry.get("type") != "puzzle": continue puzzle_id = entry.get("id", "") if instance.puzzle_manager.get_puzzle(puzzle_id): continue from src.engine.puzzle import create_puzzle_from_lorebook puzzle = create_puzzle_from_lorebook(entry) if puzzle: instance.puzzle_manager.add_puzzle(puzzle) logger.info("谜题初始化: %s (%s)", puzzle.name, puzzle_id) def build_dice_constraint_block( instance: GameInstance, actions_text: str, rule: Any, dice_system: str, dice_resolver: Any, *, planned_only: bool = False, skip_action_indexes: set[int] | None = None, ) -> str: """逐个结算玩家行动中的 CheckRequest；原始骰值只生成一次。""" if dice_system == "none": return "" blocks: list[str] = [] legacy_roll_re = re.compile(r"\(系统掷骰:\s*(d20|d100)=(\d+)\)") skipped = skip_action_indexes or set() for action_index, action in enumerate(instance.action_queue): if action_index in skipped: action.pop("check_request", None) action.pop("dice_value", None) action.pop("dice_rolls", None) action.pop("dice_system", None) action.pop("dice_roll_source", None) action.pop("dice_pending", None) continue if action.get("user_id") not in instance.players: continue request = action.get("check_request") if not isinstance(request, dict): if planned_only: continue request = build_check_request(instance, action, rule) legacy_match = legacy_roll_re.search(str(action.get("text") or "")) if not request and legacy_match: uid = str(action.get("user_id") or "") request = { "check_id": "", "required": True, "actor_uid": uid, "actor_name": instance.players.get(uid, {}).get("character_name", uid), "dice_system": legacy_match.group(1), "label": "行动检定", "intent": "legacy", "skill": str(action.get("selected_skill") or ""), "attribute": str(action.get("selected_attribute") or ""), "advantage_mode": "", "advantage_note": None, } if not request: continue action["check_request"] = request if not action.get("dice_value"): legacy_match = legacy_roll_re.search(str(action.get("text") or "")) if legacy_match: action["dice_value"] = int(legacy_match.group(2)) action["dice_rolls"] = [int(legacy_match.group(2))] action["dice_system"] = legacy_match.group(1) action["dice_roll_source"] = "legacy" else: rolled = roll_check_request(request, rule) action["dice_value"] = rolled["value"] action["dice_rolls"] = rolled["rolls"] action["dice_system"] = rolled["dice_system"] action["dice_roll_source"] = "system" action["dice_pending"] = False block = dice_resolver.resolve_action_check(instance, action, rule) if block: blocks.append(block) return "\n".join(blocks)