"""战斗结算解析器。 攻击命中只消费本轮统一检定产生的 ``CheckResult``；本模块 不存在 d20/d100 命中掷骰的兼容回退。 """ from __future__ import annotations from src.engine.modules import game_config import logging from typing import Any from src.engine.combat import AttackResult, resolve_attack, _check_succeeded from src.engine.character_utils import is_conscious, record_death_save_failures, sync_death_from_hp from src.engine.constants import WEAPON_DAMAGE, canonical_item_key from src.engine.game_instance import GameInstance from src.engine.language import localized_text from src.engine import module_hooks from src.engine.modules import checks, progression_state logger = logging.getLogger("trpg") class CombatResolver: """处理战斗目标、权威检定消费与伤害。""" @staticmethod def _authoritative_attack_check( instance: GameInstance, action: dict[str, Any], actor_uid: str, ) -> dict[str, Any] | None: """严格按 ``check_id + actor_uid + kind=attack`` 定位唯一结果。""" request = action.get("check_request") if not isinstance(request, dict): return None check_id = str(request.get("check_id") or "") if ( not check_id or str(request.get("actor_uid") or "") != actor_uid or str(request.get("kind") or "") != "attack" ): return None matches = [ check for check in checks.last_checks(instance) if str(check.get("check_id") or "") == check_id and str(check.get("actor_uid") or "") == actor_uid and str(check.get("kind") or "") == "attack" ] if len(matches) != 1: logger.warning( "攻击检定无法唯一匹配，已拒绝伤害结算: actor=%s check=%s matches=%d", actor_uid, check_id, len(matches), ) return None check = matches[0] request_opponent = str(request.get("opponent") or "") result_opponent = str(check.get("opponent") or "") if request_opponent != result_opponent: logger.warning( "攻击检定目标引用不一致，已拒绝伤害结算: check=%s request=%s result=%s", check_id, request_opponent, result_opponent, ) return None return check @staticmethod def _target_from_ref( instance: GameInstance, target_ref: str, ) -> tuple[dict[str, Any] | None, str, str]: """返回 ``(target state, display name, player uid)``。""" if target_ref in instance.players: sheet = instance.get_character_sheet(target_ref) name = str(instance.players[target_ref].get("character_name") or target_ref) return sheet, name, target_ref if target_ref.startswith("npc:"): npc_id = target_ref.split(":", 1)[1] npc = instance.npcs.get(npc_id) if isinstance(npc, dict): name = str(npc.get("character_name") or npc.get("name") or npc_id) return npc, name, "" return None, "", "" return None, "", "" @staticmethod def _weapon(character_sheet: dict[str, Any], action_text: str) -> tuple[dict[str, Any] | None, str]: equipment = character_sheet.get("equipment", []) or [] weapons: list[dict[str, Any]] = [] for item in equipment: if not isinstance(item, dict): continue name = str(item.get("name") or "").strip() explicit_type = str(item.get("type") or "").strip().casefold() key = str(item.get("item_key") or canonical_item_key(name) or "").casefold() # A non-empty type is authoritative. Only records with no type at # all use the pre-item-type name/legacy damage compatibility path. is_weapon = ( explicit_type == "weapon" if explicit_type else bool(name) and (key in WEAPON_DAMAGE or name.casefold() in WEAPON_DAMAGE) ) if is_weapon: weapons.append({**item, "name": name or "徒手", "damage": item.get("damage", 2)}) normalized_action = action_text.casefold() for weapon in weapons: name = str(weapon["name"]) key = str(weapon.get("item_key") or canonical_item_key(name) or "").casefold() if (name and name.casefold() in normalized_action) or (key and key in normalized_action): return weapon, name for weapon in weapons: if weapon.get("slot") == "main_hand": return weapon, str(weapon["name"]) if weapons: return weapons[0], str(weapons[0]["name"]) return None, "徒手" @staticmethod def _record_result(instance: GameInstance, result: AttackResult) -> None: """保持旧字段并补齐多人结算引用；同 check_id 不重复追加。""" if any( str(item.get("check_id") or "") == result.check_id for item in module_hooks.read(instance, module_hooks.COMBAT_RESULTS) ): return instance.record_combat_result({ "attacker": result.attacker, "attacker_uid": result.attacker_uid, "target": result.target, "target_ref": result.target_ref, "weapon": result.weapon, "check_id": result.check_id, "verdict": result.verdict, "is_critical": result.is_critical, "is_fumble": result.is_fumble, "damage": result.actual_damage, "actual_damage": result.actual_damage, "target_hp_before": result.target_hp_before, "target_hp_after": result.target_hp_after, "description": result.description, "round": progression_state.round_value(instance), }) def resolve_combat(self, instance: GameInstance, actions_text: str, combat_model: str, rule: Any | None = None) -> str: """只结算有唯一权威攻击 ``CheckResult`` 的行动。""" del actions_text # 禁止从多人汇总文本猜攻击者、武器或目标。 results: list[tuple[AttackResult, dict[str, Any]]] = [] for action in list(instance.action_queue): actor_uid = str(action.get("user_id") or "") if actor_uid not in instance.players or not instance.is_alive(actor_uid): continue if not is_conscious(instance.get_character_sheet(actor_uid)): continue check = self._authoritative_attack_check(instance, action, actor_uid) if check is None: continue target_ref = str(check.get("opponent") or "") target, target_name, target_uid = self._target_from_ref(instance, target_ref) if target is None: logger.warning( "攻击检定缺少可解析目标，已拒绝伤害结算: actor=%s check=%s opponent=%s", actor_uid, check.get("check_id"), target_ref, ) continue player = instance.players[actor_uid] character_sheet = instance.get_character_sheet(actor_uid) attacker_name = str(player.get("character_name") or actor_uid) weapon, weapon_name = self._weapon(character_sheet, str(action.get("text") or "")) request = action.get("check_request") if isinstance(action.get("check_request"), dict) else {} attribute_key = str(check.get("attribute_key") or request.get("attribute") or "str").strip().casefold() attr_value = int(character_sheet.get("attributes", {}).get(attribute_key, 10) or 10) was_unconscious = str((target or {}).get("status") or "") in {"downed", "stable"} attacker_faction = str(character_sheet.get("faction") or "party") target_faction = "" if target_uid: target_faction = str(instance.get_character_sheet(target_uid).get("faction") or "party") same_faction = bool(attacker_faction and target_faction and attacker_faction == target_faction) cached = action.get("combat_outcome") if isinstance(cached, dict): try: result = AttackResult.from_record(cached) except (TypeError, ValueError): logger.warning( "攻击缓存格式无效，已拒绝重放: actor=%s check=%s", actor_uid, check.get("check_id"), ) continue if ( result.check_id != str(check.get("check_id") or "") or result.attacker_uid != actor_uid or result.target_ref != target_ref ): logger.warning( "攻击缓存引用不一致，已拒绝重放: actor=%s check=%s", actor_uid, check.get("check_id"), ) continue else: result = resolve_attack( attacker_name=attacker_name, target=target, weapon=weapon, attr_value=attr_value, combat_model=combat_model, difficulty=game_config.difficulty(instance), check_result=check, same_faction=same_faction, attacker_uid=actor_uid, target_ref=target_ref, target_name=target_name, rule=rule, ) # 声明昏迷机制的规则：玩家目标 HP 归零立即落昏迷（旧版即死语义不变，仍走状态更新路径）。 if ( target_uid and rule is not None and rule.death_mechanic["hp_zero"] == "downed_death_saves" ): sync_death_from_hp(target, progression_state.round_value(instance), rule) # 5e：昏迷中受击累加死亡豁免失败（任何伤害 1 次、暴击 2 次）。 if ( was_unconscious and rule is not None and rule.death_mechanic["hp_zero"] == "downed_death_saves" and _check_succeeded(check) and result.actual_damage > 0 ): failures = 2 if result.is_critical else 1 outcome = record_death_save_failures( target, failures, progression_state.round_value(instance) ) result.description += localized_text(instance.language, { "zh-CN": ( "（昏迷受击，死亡豁免失败满3次，死亡）" if outcome == "dead" else f"（昏迷受击，死亡豁免失败+{failures}）" ), "en": ( " (hit while unconscious: 3rd failure, dies)" if outcome == "dead" else f" (hit while unconscious: death save failures +{failures})" ), "ja": ( "（意識不明で被弾：失敗3回で死亡）" if outcome == "dead" else f"（意識不明で被弾：死亡セーヴ失敗+{failures}）" ), "de": ( " (bewusstlos getroffen: 3. Fehlschlag, gestorben)" if outcome == "dead" else f" (bewusstlos getroffen: Rettungswurf-Fehlschläge +{failures})" ), }) # 战斗 outcome 属于行动的下游状态，不回写或改动 # 已形成的 CheckResult。进程重试时复用它，既不重掷 # 命中骰，也不重复扣 HP。 action["combat_outcome"] = result.to_record() # 旧记录中可能没有 weapon；仅在内存恢复时填上显示值。 if not result.weapon: result.weapon = weapon_name self._record_result(instance, result) results.append((result, check)) if not results: return "" lines = [localized_text(instance.language, { "zh-CN": "【系统战斗结算·必须遵循】", "en": "[System Combat Resolution - Must Follow]", "ja": "【システム戦闘結算・必ず従うこと】", "de": "[Systemseitige Kampfabrechnung - Muss befolgt werden]", })] for result, check in results: lines.append(localized_text(instance.language, { "zh-CN": f"{result.attacker}持{result.weapon}攻击{result.target}", "en": f"{result.attacker} attacks {result.target} with {result.weapon}", "ja": f"{result.attacker}は{result.weapon}で{result.target}を攻撃した", "de": f"{result.attacker} greift {result.target} mit {result.weapon} an", })) dice_name = str(check.get("dice") or "") roll_value = int(check.get("roll", 0) or 0) lines.append(localized_text(instance.language, { "zh-CN": ( f" {dice_name}={roll_value} → {result.verdict}，" f"最终伤害={result.actual_damage}" ), "en": ( f" {dice_name}={roll_value} → {result.verdict}, " f"final damage={result.actual_damage}" ), "ja": ( f" {dice_name}={roll_value} → {result.verdict}、" f"最終ダメージ={result.actual_damage}" ), "de": ( f" {dice_name}={roll_value} → {result.verdict}, " f"Endschaden={result.actual_damage}" ), })) if result.is_critical: lines.append(localized_text(instance.language, { "zh-CN": " ⚡ 大成功！", "en": " ⚡ Critical!", "ja": " ⚡ 大成功！", "de": " ⚡ Kritischer Erfolg!", })) if result.target_hp_after <= 0 and result.target_hp_before > 0: lines.append(localized_text(instance.language, { "zh-CN": f" 💀 {result.target} 倒地！", "en": f" 💀 {result.target} is down!", "ja": f" 💀 {result.target} は倒れた！", "de": f" 💀 {result.target} ist niedergestreckt!", })) logger.info( "多人战斗结算: %d attacks, checks=%s", len(results), [result.check_id for result, _ in results], ) return "\n".join(lines)