src/webui/services/characters.py
Rama main · réplica de lectura del repositorio original.
"""角色管理服务：角色列表 / 规则属性辅助 / 角色CRUD / 建卡。"""
from __future__ import annotations
from src.engine.modules import game_config
import logging
import re
import time
from collections.abc import Awaitable, Callable
from copy import deepcopy
from dataclasses import dataclass
from pathlib import Path
from typing import TYPE_CHECKING, Any, cast
from uuid import uuid4
from src.engine.character_utils import (
build_starter_items,
calc_hp_from_rule,
initial_special_stat_value,
make_default_character,
normalize_character_sheet,
set_hp,
)
from src.compat.characters import MAX_SKILL_EFFECT_CHARS
from src.content.worlds import localize_lorebook_entries
from src.engine.language import localized_text
from src.engine.health import record_health_event
from src.engine.dice_rng import parse_dice_formula
from src.engine.economy import (
MAX_ECONOMY_AMOUNT,
complete_effect_group,
queue_proposal,
queue_purchase_offer,
resolve_proposal,
cancel_proposals_for_player,
)
from src.engine.memory_outbox import (
complete_memory_delivery,
complete_memory_reversal,
pending_memory_deliveries,
pending_memory_reversals,
queue_memory_delivery,
)
from src.engine.game_instance import GameInstance
from src.engine import module_hooks
from src.engine.modules import progression_state, room_access, ruleset_runtime, skill_growth
from src.engine.player_control import claim_seat, is_human_controlled
from src.content_modules.projection import ContentProjectionService, entries_for_world
from src.commands.economy_effects import pending_decision_notice
from src.commands.state_items import grant_classified_item
from src.rulesets.contracts import GameDetailProjectionRuntime, PlayerJoinRuntime
from src.webui.character_contracts import MAX_BIO_CHARS
from src.webui.services._common import professional_character_validation_error
from src.webui.character_sheet_authority import (
EDITABLE_SHEET_FIELDS,
adopt_requires_gm_failure,
delete_requires_gm_failure,
field_requires_gm_failure,
player_field_violations,
)
if TYPE_CHECKING:
from src.rulesets.registry import RulesetRuntimeRegistry
logger = logging.getLogger("trpg")
def _validate_damage_dice(equipment: Any) -> str | None:
"""Validate an optional damage formula without changing rule mechanics.
``damage`` remains the fixed fallback. ``damage_dice`` uses the shared
generic parser (NdM with an optional +/-K modifier); the field is merely
preserved on the generic character sheet and is consumed only by rules
that already declare dice damage.
"""
if not isinstance(equipment, list):
return None
for item in equipment:
if not isinstance(item, dict):
continue
raw_formula = item.get("damage_dice")
if raw_formula is None or (isinstance(raw_formula, str) and not raw_formula.strip()):
continue
if not isinstance(raw_formula, str):
return f"装备 {item.get('name', '?')} 的伤害骰公式无效: 必须是文本"
try:
parse_dice_formula(raw_formula)
except ValueError as exc:
return f"装备 {item.get('name', '?')} 的伤害骰公式无效: {exc}"
return None
_TRAILING_SENTENCE_PUNCTUATION = "。．.！!？?；;，,、"
def _settlement_reason_text(reason: str) -> str:
"""Strip trailing sentence punctuation so the composed card never shows
a doubled full stop（提案 reason 由模型书写，常自带句号）。"""
return str(reason or "").rstrip().rstrip(_TRAILING_SENTENCE_PUNCTUATION).rstrip()
def _record_economy_outcome_in_round(
instance: Any,
outcome: dict[str, Any],
) -> None:
"""Attach the authoritative decision to its originating public round."""
status = str(outcome.get("status") or "")
effects_status = str(outcome.get("effects_status") or "none")
amount = int(outcome.get("amount", 0) or 0)
reason = _settlement_reason_text(str(outcome.get("reason") or "经济提案"))
if status == "committed" and effects_status == "pending":
message = localized_text(instance.language, {
"en": f"Settlement confirmed ({amount}): {reason}. Linked results are waiting for the remaining decisions.",
"zh-CN": f"结算已确认（{amount}）：{reason}。关联结果仍在等待其余决定。",
"ja": f"決済確認済み（{amount}）：{reason}。関連結果は残りの判断を待っています。",
"de": f"Abrechnung bestätigt ({amount}): {reason}. Verknüpfte Ergebnisse warten noch auf die restlichen Entscheidungen.",
})
elif status == "committed":
message = localized_text(instance.language, {
"en": f"Settlement confirmed ({amount}): {reason}. Dependent results are now effective.",
"zh-CN": f"结算已确认（{amount}）：{reason}。关联结果现已生效。",
"ja": f"決済確認済み（{amount}）：{reason}。関連結果が発効しました。",
"de": f"Abrechnung bestätigt ({amount}): {reason}. Abhängige Ergebnisse sind jetzt wirksam.",
})
else:
message = localized_text(instance.language, {
"en": f"Settlement {status} ({amount}): {reason}. No payment or dependent result occurred.",
"zh-CN": f"结算未成立（{amount}）：{reason}。没有付款，关联结果也未生效。",
"ja": f"決済不成立（{amount}）：{reason}。支払いも関連結果も発生していません。",
"de": f"Abrechnung {status} ({amount}): {reason}. Es erfolgte weder eine Zahlung noch ein abhängiges Ergebnis.",
})
round_number = int(outcome.get("round", 0) or 0)
# Purchases are always a party-visible settlement event. Keep the
# explicit private path for other GM payments, but also promote purchases
# created before the visibility default was corrected. Solo tables have
# no information asymmetry: the only player is the GM, so settlement
# messages belong on the shared timeline.
is_purchase = str(outcome.get("kind") or "") == "purchase"
if (
str(outcome.get("visibility") or "private") != "party"
and not is_purchase
and not bool(game_config.solo_mode(instance))
):
recipients = {
str(outcome.get("payer_uid") or ""),
str(outcome.get("recipient_uid") or ""),
}
for uid in recipients:
if uid in instance.players:
instance.append_private_message(uid, {
"round": round_number,
"text": message,
"kind": "economy_resolution",
})
return
entry = next(
(
item for item in reversed(instance.log)
if int(item.get("round", -1) or -1) == round_number
),
None,
)
if entry is None:
return
resolutions = entry.setdefault("economy_resolutions", [])
outcome_id = str(outcome.get("id") or "")
if not any(str(item.get("id") or "") == outcome_id for item in resolutions):
resolutions.append(dict(outcome))
changes = entry.setdefault("state_changes", [])
if message not in changes:
changes.append(message)
def _remove_stale_pending_notice(instance: Any) -> None:
"""Drop the round's "settlement pending" notice once nothing is pending.
The notice is only accurate while the response's proposals and effect
groups are unresolved; once everything is settled, declined, or
discarded, it must not keep telling the table that dependent results are
still pending (real incident: a sale settled but the round kept showing
the pending banner as its only status line).
"""
from src.engine.economy import has_pending_economy_decision
if has_pending_economy_decision(instance):
return
notice = pending_decision_notice(str(getattr(instance, "language", "") or ""))
for entry in instance.log:
changes = entry.get("state_changes")
if isinstance(changes, list) and notice in changes:
changes.remove(notice)
@dataclass(frozen=True)
class CharacterGameDependencies:
get_instance: Callable[[tuple[str, ...]], GameInstance | None]
parse_game_key: Callable[[str], tuple[str, ...]]
save_instance: Callable[[GameInstance], Awaitable[None]]
@dataclass(frozen=True)
class CharacterRuleDependencies:
rules_dir: Path | None
load_rule_by_id: Callable[[str, str], Any | None]
load_rule_for_game: Callable[[Any], Any | None]
ruleset_registry: "RulesetRuntimeRegistry"
@dataclass(frozen=True)
class CharacterAssetDependencies:
lorebook: Any | None
load_world_template: Callable[[str, str], dict[str, Any] | None]
avatar_file: Callable[[str], Path | None]
generated_image_file: Callable[[str], Path | None]
content_projection: ContentProjectionService | None = None
# GAL P8：立绘资产边界校验（与 avatar_file 同语义）。缺省 None 时立绘
# 引用一律拒绝（fail closed），不猜文件是否存在。
standing_portrait_file: Callable[[str], Path | None] | None = None
@dataclass(frozen=True)
class CharacterDependencies:
games: CharacterGameDependencies
rules: CharacterRuleDependencies
assets: CharacterAssetDependencies
save_character_card: Callable[[dict[str, Any]], dict[str, Any]]
apply_economy_effects: Callable[[Any, dict[str, Any]], Awaitable[None]] | None = None
schedule_economy_scene_image: Callable[[Any, dict[str, Any]], Any] | None = None
apply_economy_memory: Callable[[str, str, dict[str, Any], int], Awaitable[None]] | None = None
reverse_economy_memory: Callable[[str, str], Awaitable[bool]] | None = None
# GAL P8：个人音色存在性检查（speech 的 profile store）。投影用它把已删除
# 的音色绑定省略掉（失效引用保持未配置）；None 时投影原样带出，
# 合成端仍会 fail closed 明确报错。
voice_profile_exists: Callable[[str], bool] | None = None
async def create_payment_proposal(
dependencies: CharacterDependencies,
game_key: str,
*,
payer_uid: str,
amount: int,
reason: str = "",
recipient_uid: str = "",
items: list[str] | None = None,
) -> dict[str, Any]:
"""Create a GM-authored payment proposal without changing balances."""
inst = dependencies.games.get_instance(
dependencies.games.parse_game_key(game_key),
)
if inst is None:
return {"ok": False, "error": "游戏不存在"}
payer_uid = str(payer_uid or "").strip()
recipient_uid = str(recipient_uid or payer_uid).strip()
try:
amount = int(amount)
except (TypeError, ValueError):
return {"ok": False, "error": "金额必须是整数"}
if not 0 < amount <= MAX_ECONOMY_AMOUNT:
return {"ok": False, "error": f"金额必须在 1 到 {MAX_ECONOMY_AMOUNT} 之间"}
if payer_uid not in inst.players:
return {"ok": False, "error": "付款角色不存在"}
if recipient_uid not in inst.players:
return {"ok": False, "error": "接收角色不存在"}
normalized_items = [
str(item).strip()[:120]
for item in list(items or [])[:8]
if str(item).strip()
]
reason = str(reason or "").strip()[:240]
if not reason:
reason = f"购买 {'、'.join(normalized_items)}" if normalized_items else "GM 发起的支付"
async with inst.authoritative_write() as write_entered, inst._lock:
if not write_entered:
return {
"ok": False,
"code": "REWRITE_IN_PROGRESS",
"error": "GM 正在重写历史回合，请稍后再发起支付",
}
current = dependencies.games.get_instance(inst.game_key)
if current is None or current is not inst:
return {
"ok": False,
"code": "STALE_RUN",
"error": "游戏已重开或重置，请刷新后再发起支付",
}
if normalized_items:
try:
proposal = queue_purchase_offer(
inst,
payer_uid=payer_uid,
amount=amount,
items=normalized_items,
reason=reason,
source="gm_manual",
source_ref=f"gm_manual:{inst.run_id}:{uuid4().hex}",
)
except ValueError as exc:
return {"ok": False, "code": "ORDER_INVALID", "error": str(exc)}
else:
proposal = queue_proposal(
inst,
kind="payment",
payer_uid=payer_uid,
recipient_uid=recipient_uid,
amount=amount,
rewards=[],
reason=reason,
source="gm_manual",
source_ref=f"gm_manual:{inst.run_id}:{uuid4().hex}",
approval_policy="payer",
visibility="party",
)
await dependencies.games.save_instance(inst)
return {"ok": True, "proposal": proposal}
_ATTR_NAME_EN = {
"str": "STR",
"con": "CON",
"dex": "DEX",
"int": "INT",
"edu": "EDU",
"app": "APP",
"pow": "POW",
"siz": "SIZ",
"wis": "WIS",
"cha": "CHA",
}
_ATTR_NAME_ZH = {
"str": "力量",
"con": "体质",
"dex": "敏捷",
"int": "智力",
"edu": "教育",
"app": "外貌",
"pow": "意志",
"siz": "体型",
"wis": "感知",
"cha": "魅力",
}
def _creation_bounds_enabled(rule) -> bool:
policy = rule.template.get("character_creation") if rule else None
return isinstance(policy, dict) and policy.get("clamp_to_rule_bounds") is True
def creation_bounds_violation(rule: Any, character_sheet: dict[str, Any]) -> dict[str, Any] | None:
"""规则声明了服务端建卡边界时，返回拒绝响应；通过则返回 None。
属性 clamp 只是第一道：技能点预算（含 CoC 的 above_base 公式）、技能条数、
单技能上限、装备品质都必须在这里按同一套规则再校验一次，否则绕过前端直接调
API 就能把超预算角色带进局内。只对显式声明 ``character_creation.clamp_to_rule_bounds``
的规则生效，避免把 CoC 的限制推广到 D&D / Freeform。
中性/旧式部分角色卡（属性缺失或只有个别属性）不在此校验：``validate_character``
把缺失属性当 0，会把中性冒险入场、旧卡迁移这类合法形态误判为非法——它们继续
走既有的 clamp/normalize 兜底（与修复前行为一致）。只要角色卡带齐了规则声明的
全部属性，就视为正式建卡结果，执行全量校验。
"""
if rule is None:
return None
# 只有「带齐规则声明的全部属性」的卡才视为正式建卡结果；属性缺失/为空/部分
# （中性冒险入场、旧卡迁移）一律走既有 clamp/normalize 兜底，不在此校验。
attributes = character_sheet.get("attributes")
declared_keys = [str(spec.get("key") or "") for spec in rule.attributes if spec.get("key")]
if not isinstance(attributes, dict) or any(key not in attributes for key in declared_keys):
return None
violations = rule.validate_character(character_sheet)
if not violations:
return None
return {
"ok": False,
"error_code": "CHARACTER_SHEET_INVALID",
"error": "角色卡不符合当前规则：" + "；".join(str(item) for item in violations[:3]),
}
def _clamp_attributes(rule, attributes: Any) -> dict:
"""入场时把属性压回规则声明的上下限：客户端不能自报超出范围的属性。"""
result = dict(attributes) if isinstance(attributes, dict) else {}
for spec in rule.attributes:
key = str(spec.get("key") or "")
if not key or key not in result:
continue
try:
value = int(result[key])
except (TypeError, ValueError):
value = int(spec.get("min", 0) or 0)
low, high = spec.get("min"), spec.get("max")
if isinstance(low, int):
value = max(low, value)
if isinstance(high, int):
value = min(high, value)
result[key] = value
return result
def _normalize_skills(skills: list, rule=None, *, attributes: Any = None, clamp: bool = False) -> list[dict]:
"""规范化技能列表：字符串转为含数值的对象格式，并保留可选 effect 说明。
「超出基础值才计点」的规则（CoC）缺省数值取该角色的技能基础值（含闪避、
母语等由属性派生的基础值）；其它规则保持旧的缺省 20。``clamp`` 为真时
单项技能不超过规则的上限。
"""
base_values: dict[str, int] = rule.skill_base_values if rule else {}
above_base = bool(rule and rule.skill_point_spend_mode == "above_base")
cap = int(rule.max_skill_value or 0) if (rule and clamp) else 0
def default_value(name: str) -> int:
return rule.skill_base_value_for(name, attributes) if above_base else base_values.get(name, 20)
def bounded(value: Any) -> Any:
if not cap:
return value
try:
return max(0, min(cap, int(value)))
except (TypeError, ValueError):
return 0
result: list[dict] = []
for s in skills:
if isinstance(s, str):
result.append({"name": s, "value": bounded(default_value(s))})
elif isinstance(s, dict):
name = s.get("name", "")
row: dict = {
"name": name,
"value": bounded(s.get("value", default_value(name))),
}
effect = str(s.get("effect") or "").strip()
if effect:
row["effect"] = effect[:MAX_SKILL_EFFECT_CHARS]
result.append(row)
return result
def _format_rule_attr(attr: dict) -> dict:
key = attr["key"]
name = attr.get("name") or _ATTR_NAME_ZH.get(key, key)
name_en = attr.get("name_en") or _ATTR_NAME_EN.get(key, key.upper())
return {
"key": key,
"name": name,
"name_en": name_en,
"display_name": attr.get("display_name") or (f"{name} ({name_en})" if name_en else name),
"min": attr.get("min", 3),
"max": attr.get("max", 18),
}
def _fallback_rule_attrs() -> list[dict[str, Any]]:
return [
_format_rule_attr({"key": key, "name": name, "min": 3, "max": 18})
for key, name in [
("str", "力量"), ("dex", "敏捷"), ("con", "体质"),
("int", "智力"), ("wis", "感知"), ("cha", "魅力"),
]
]
def _fallback_rule_meta() -> dict[str, Any]:
return {
"dice_system": "d20",
"rule_id": "",
"mechanics": "freeform_d20_core",
"hp_formula": "",
"auto_hp": False,
"attr_hint": "",
"skill_mode": "narrative",
"skill_hint": "",
"max_skills": 3,
"skill_point_total": 0,
"max_skill_value": 0,
"skill_point_spend_mode": "total_value",
"skill_pools": {},
"skill_base_values": {},
"currency": "金币",
"conflict_model": {"type": "hp_based"},
"currency_system": {"base_unit": "unit", "units": [{"id": "unit", "name": "金币", "rate": 1}]},
"resource_schema": [{"key": "hp", "label": "生命", "min": 0}],
"identity_schema": [
{"key": "origin", "label": "种族", "type": "text", "legacy_field": "race"},
{"key": "archetype", "label": "职业", "type": "text", "legacy_field": "class"},
{"key": "background", "label": "背景", "type": "text", "legacy_field": "background"},
],
"progression_schema": {"type": "xp_level"},
"ui_schema": {
"primary_resources": ["hp"],
"secondary_resources": [],
"identity_labels": {"origin": "种族", "archetype": "职业", "background": "背景"},
"show_level": True,
"show_xp": True,
"currency_label": "金币",
"equipment_label": "装备",
},
}
def _character_schema_for_rule(rule) -> dict[str, Any]:
"""Build the one character-creation contract used with and without a game."""
if not rule:
return {
"rule_attrs": _fallback_rule_attrs(),
"rule_attrs_total": 60,
"rule_classes": ["战士", "法师", "游侠", "盗贼", "牧师", "冒险者"],
"rule_special_stats": [],
"rule_meta": _fallback_rule_meta(),
"skill_pool": [],
}
meta = {
"dice_system": rule.dice_system,
"rule_id": rule.rule_id,
"rule_name": rule.rule_name,
"rule_version": str(rule.template.get("rule_version") or ""),
"mechanics": rule.mechanics,
"hp_formula": rule.hp_formula,
"auto_hp": rule.mechanics == "coc7e_core",
"attr_hint": rule.attr_hint,
"skill_mode": rule.skill_mode,
"skill_hint": rule.skill_hint,
"max_skills": rule.max_skills,
"skill_point_total": rule.skill_point_total,
"skill_point_formula": rule.skill_point_formula,
"skill_base_lookup": rule.skill_base_lookup(),
"custom_skill_base": 5,
"max_skill_value": rule.max_skill_value,
"skill_point_spend_mode": rule.skill_point_spend_mode,
"skill_pools": rule.skill_pools,
"skill_base_values": rule.skill_base_values,
"currency": rule.currency,
"conflict_model": rule.conflict_model,
"currency_system": rule.currency_system,
"resource_schema": rule.resource_schema,
"identity_schema": rule.identity_schema,
"progression_schema": rule.progression_schema,
"ui_schema": rule.ui_schema,
}
explicit_skill_pool = rule.template.get("skill_pool")
legacy_skills = rule.template.get("skills")
if isinstance(explicit_skill_pool, list):
skill_pool = list(explicit_skill_pool)
elif isinstance(legacy_skills, list):
skill_pool = list(legacy_skills)
else:
skill_pool = []
if not skill_pool:
# Standalone creation has no class selection context yet. Offer the
# union of class pools so the same rule schema remains useful before a
# game (and before the user has settled on an archetype).
for class_pool in rule.skill_pools.values():
for skill in class_pool:
if skill not in skill_pool:
skill_pool.append(skill)
return {
"rule_attrs": [_format_rule_attr(attr) for attr in rule.attributes],
"rule_attrs_total": rule.attribute_points,
"rule_classes": rule.get_class_names(),
"rule_special_stats": rule.special_stats,
"rule_meta": meta,
"skill_pool": skill_pool,
}
def character_schema(
dependencies: CharacterDependencies,
rule_id: str,
language: str = "",
) -> dict[str, Any]:
"""Return rule-driven creation fields without requiring a GameInstance."""
rule = dependencies.rules.load_rule_by_id(rule_id, language)
if not rule:
return {"ok": False, "error": f"规则不存在: {rule_id}"}
return {
"ok": True,
**_character_schema_for_rule(rule),
"ruleset_runtime": dependencies.rules.ruleset_registry.describe(
rule.template,
).to_dict(),
}
def format_attribute_map(attributes: dict, rule_attrs: list[dict]) -> str:
"""按规则属性顺序格式化属性，中文界面同时显示中文名与英文 key。"""
attr_by_key = {a["key"]: a for a in rule_attrs}
keys = [a["key"] for a in rule_attrs]
keys.extend(k for k in attributes if k not in attr_by_key)
parts = []
for key in keys:
if key not in attributes:
continue
attr = _format_rule_attr(attr_by_key.get(key) or {"key": key})
parts.append(f"{attr['display_name']}:{attributes[key]}")
return " ".join(parts)
def list_characters(
dependencies: CharacterDependencies,
game_key: str,
*,
viewer_is_gm: bool,
) -> dict[str, Any]:
inst = dependencies.games.get_instance(
dependencies.games.parse_game_key(game_key),
)
if not inst:
return {"players": [], "npcs": [], "rule_attrs": []}
players = [
# ``has_acted``: after its first action a seat's card / deletion is GM-only.
{"user_id": uid, **p, "has_acted": module_hooks.seat_has_acted(inst, uid)}
for uid, p in inst.players.items()
]
rule_attrs = _get_rule_attrs_for_game(dependencies, inst)
rule = dependencies.rules.load_rule_for_game(inst)
for player in players:
cs = player.get("character_sheet", {})
normalize_character_sheet(cs, rule)
cs["attributes_display"] = format_attribute_map(cs.get("attributes", {}), rule_attrs)
npcs_by_name: dict[str, dict[str, Any]] = {}
if viewer_is_gm:
for nid, npc in inst.npcs.items():
name = npc.get("character_name") or npc.get("name") or nid
row = {"npc_id": nid, **npc, "name": name}
_project_npc_bindings(row, dependencies)
npcs_by_name[name] = row
if dependencies.assets.lorebook and inst.world_id:
if dependencies.assets.content_projection is not None:
entries = dependencies.assets.content_projection.for_game(
inst, viewer_kind="gm", entry_type="npc",
)
else:
entries = entries_for_world(dependencies.assets.lorebook, inst.world_id, "npc")
world_data = dependencies.assets.load_world_template(
inst.world_id,
str(getattr(inst, "language", "") or ""),
)
lore_status = localized_text(
getattr(inst, "language", ""),
{"en": "Lorebook", "zh-CN": "世界书", "ja": "ワールドブック", "de": "Lorebook"},
)
for entry in localize_lorebook_entries(entries, world_data):
name = entry.get("name", "")
if not name or name in npcs_by_name:
continue
npcs_by_name[name] = {
"npc_id": entry.get("id", name),
"name": name,
"character_name": name,
"tier": entry.get("tier", ""),
"status": lore_status,
"relation": entry.get("relation", ""),
"content": entry.get("content", ""),
"portrait": entry.get("portrait"),
}
npcs = list(npcs_by_name.values())
rule_attrs_total = _get_rule_attrs_total(dependencies, inst)
result: dict[str, Any] = {
"players": players,
"npcs": npcs,
"rule_attrs": rule_attrs,
"rule_attrs_total": rule_attrs_total,
"rule_classes": _get_rule_classes_for_game(dependencies, inst),
"rule_special_stats": _get_rule_special_stats(dependencies, inst),
"rule_meta": _get_rule_meta_for_game(dependencies, inst),
}
if rule is not None:
runtime_metadata = dependencies.rules.ruleset_registry.describe(
rule.template,
).to_dict()
result["ruleset_runtime"] = runtime_metadata
runtime = dependencies.rules.ruleset_registry.resolve(rule.template)
if isinstance(runtime, GameDetailProjectionRuntime):
result.update(runtime.game_detail_projection(inst))
return result
def _get_rule_classes_for_game(
dependencies: CharacterDependencies,
inst: Any,
) -> list[str]:
try:
rule = dependencies.rules.load_rule_for_game(inst)
if rule:
return _character_schema_for_rule(rule)["rule_classes"]
except Exception:
logger.exception("读取规则职业失败: world_id=%s", inst.world_id)
return ["战士", "法师", "游侠", "盗贼", "牧师", "冒险者"]
def _get_rule_attrs_for_game(
dependencies: CharacterDependencies,
inst: Any,
) -> list[dict]:
try:
rule = dependencies.rules.load_rule_for_game(inst)
if rule:
return _character_schema_for_rule(rule)["rule_attrs"]
except Exception:
logger.exception("读取规则属性失败: world_id=%s", inst.world_id)
return _fallback_rule_attrs()
def _get_rule_attrs_total(dependencies: CharacterDependencies, inst: Any) -> int:
try:
rule = dependencies.rules.load_rule_for_game(inst)
if rule:
return _character_schema_for_rule(rule)["rule_attrs_total"]
except Exception:
logger.exception("读取规则属性点失败: world_id=%s", inst.world_id)
return 60
def _get_rule_special_stats(
dependencies: CharacterDependencies,
inst: Any,
) -> list[dict]:
try:
rule = dependencies.rules.load_rule_for_game(inst)
if rule:
return _character_schema_for_rule(rule)["rule_special_stats"]
except Exception:
logger.exception("读取特殊属性失败: world_id=%s", inst.world_id)
return []
def _get_rule_meta_for_game(
dependencies: CharacterDependencies,
inst: Any,
) -> dict[str, Any]:
try:
rule = dependencies.rules.load_rule_for_game(inst)
if rule:
return _character_schema_for_rule(rule)["rule_meta"]
except Exception:
logger.exception("读取规则建卡提示失败: world_id=%s", inst.world_id)
return _fallback_rule_meta()
def get_character(
dependencies: CharacterDependencies,
game_key: str,
user_id: str,
) -> dict[str, Any] | None:
inst = dependencies.games.get_instance(
dependencies.games.parse_game_key(game_key),
)
if not inst or user_id not in inst.players:
return None
player = inst.get_player(user_id) or {}
character_sheet = player.get("character_sheet", {})
normalize_character_sheet(
character_sheet, dependencies.rules.load_rule_for_game(inst),
)
inst.set_character_sheet(user_id, character_sheet)
return {"user_id": user_id, **(inst.get_player(user_id) or {})}
async def update_character(
dependencies: CharacterDependencies,
game_key: str,
user_id: str,
updates: dict,
*,
gm_authority: bool = False,
) -> dict[str, Any]:
"""Patch a classic character sheet.
``gm_authority`` is the caller's right to change mechanics (the table GM
or the owner acting as itself). Without it only profile fields and a
level-up point allocation are accepted (see ``character_sheet_authority``).
"""
inst = dependencies.games.get_instance(
dependencies.games.parse_game_key(game_key),
)
if not inst:
return {"ok": False, "error": "角色不存在"}
async with inst.authoritative_write() as write_entered:
if not write_entered:
return {
"ok": False, "error_code": "REWRITE_IN_PROGRESS",
"error": "GM 正在重写历史回合，请等待完成后重试",
}
if dependencies.games.get_instance(inst.game_key) is not inst:
return {"ok": False, "code": "STALE_RUN", "error": "对局已重开，请刷新后重试"}
return await _update_character_authority(
dependencies, inst, user_id, updates, gm_authority=gm_authority,
)
async def adopt_library_card(
dependencies: CharacterDependencies,
game_key: str,
user_id: str,
card: dict[str, Any],
*,
gm_authority: bool,
) -> dict[str, Any]:
"""Replace a classic character from a server-owned library card.
The card comes from the library, not from the request, so it is applied
with full sheet authority. Who may adopt is still the caller's
``gm_authority``: once the seat has acted (``seat_activity``) a
player-side caller may not swap its sheet (a free refill / re-roll); only
the GM may. Before that -- a fresh seat, or one just claimed from the GM
or the AI -- the player may still switch to a card. The check runs inside
the authoritative write so it cannot race the seat's first action.
A rules-aware game answers ``RULESET_CHARACTER_OPERATION_REQUIRED`` and
the caller uses the ruleset adoption instead.
"""
inst = dependencies.games.get_instance(
dependencies.games.parse_game_key(game_key),
)
if not inst:
return {"ok": False, "error": "角色不存在"}
instance: GameInstance = inst
async with instance.authoritative_write() as write_entered:
if not write_entered:
return {
"ok": False, "error_code": "REWRITE_IN_PROGRESS",
"error": "GM 正在重写历史回合，请等待完成后重试",
}
player_side_after_acting = (
not gm_authority
and user_id in instance.players
and not _is_rules_aware_game(dependencies, instance)
and module_hooks.seat_has_acted(instance, user_id)
)
if dependencies.games.get_instance(instance.game_key) is not instance:
return {"ok": False, "code": "STALE_RUN", "error": "对局已重开，请刷新后重试"}
if player_side_after_acting:
return adopt_requires_gm_failure()
card = deepcopy(card)
for skill in card.get('skills') or []:
if isinstance(skill, dict):
skill.pop('id', None)
return await _update_character_authority(
dependencies, instance, user_id, card, gm_authority=True,
)
def _is_rules_aware_game(dependencies: CharacterDependencies, inst: GameInstance) -> bool:
rule = dependencies.rules.load_rule_for_game(inst)
if rule is None:
return False
runtime = dependencies.rules.ruleset_registry.resolve(rule.template)
return getattr(runtime.capabilities, "character_lifecycle", "legacy") == "rules_aware"
def _special_stat_fields(dependencies: CharacterDependencies, inst: GameInstance) -> set[str]:
"""Rule-defined special stats (sanity, luck...) and their maxima."""
rule = dependencies.rules.load_rule_for_game(inst)
fields: set[str] = set()
for stat in getattr(rule, "special_stats", None) or []:
key = str(stat.get("key") or "") if isinstance(stat, dict) else ""
if key:
fields.update({key, f"max_{key}"})
return fields
async def _update_character_authority(
dependencies: CharacterDependencies,
instance: GameInstance,
user_id: str,
updates: dict,
*,
gm_authority: bool = False,
) -> dict[str, Any]:
inst = instance
if not inst or user_id not in inst.players:
return {"ok": False, "error": "角色不存在"}
if _is_rules_aware_game(dependencies, inst):
return {
"ok": False,
"error_code": "RULESET_CHARACTER_OPERATION_REQUIRED",
"error": "专业规则角色不能使用旧版通用编辑接口",
}
# Only player-editable sheet fields; anything else (card metadata, plugin
# provenance, server-owned state) is dropped instead of mass-assigned.
updates = {
key: value for key, value in dict(updates).items()
if key in EDITABLE_SHEET_FIELDS or key in _special_stat_fields(dependencies, inst)
}
if not gm_authority:
denied = player_field_violations(
inst.get_character_sheet(user_id), updates,
rule_attrs=_get_rule_attrs_for_game(dependencies, inst),
)
if denied:
return field_requires_gm_failure(denied)
if "skills" in updates:
if not isinstance(updates["skills"], list):
return {"ok": False, "error": "Skills must be a list"}
old_skills = inst.get_character_sheet(user_id).get("skills", [])
accepted = set()
for row in updates["skills"]:
if not isinstance(row, dict) or not row.get("id"):
continue
sid = row['id']
matches = [s for s in old_skills if isinstance(s, dict) and s.get('id') == sid]
if (not isinstance(sid, str) or len(matches) != 1 or sid in accepted
or matches[0].get('name') != row.get('name')):
return {"ok": False, "error": "Skill identity changed or is ambiguous; reload the character and retry"}
accepted.add(sid)
pending = {mark['skill_id'] for mark in skill_growth.view(inst)['pending'] if mark['actor_uid'] == user_id}
if not pending.issubset(accepted):
return {"ok": False, "error": "A marked skill cannot be removed or renamed before settlement"}
character_name = str(updates.pop("character_name", "")).strip()
if character_name:
inst.set_player_name(user_id, character_name)
cs = inst.get_character_sheet(user_id)
if "background" in updates and len(str(updates.get("background", ""))) > MAX_BIO_CHARS:
return {"ok": False, "error": f"角色背景过长（上限 {MAX_BIO_CHARS} 字）"}
rule = dependencies.rules.load_rule_for_game(inst)
damage_dice_error = _validate_damage_dice(updates.get("equipment"))
if damage_dice_error:
return {"ok": False, "error": damage_dice_error}
if "portrait" in updates:
try:
portrait = _validated_portrait(
dependencies.assets, updates.pop("portrait"),
)
except ValueError as exc:
return {"ok": False, "error": str(exc)}
if portrait is None:
cs.pop("portrait", None)
else:
cs["portrait"] = portrait
# GAL P8：立绘 / 表情差分 / 音色绑定（全部可选字段，旧存档照常）。
standing_portrait_present = "standing_portrait" in updates
standing_expressions_present = "standing_portrait_expressions" in updates
if standing_portrait_present or standing_expressions_present:
try:
# A missing PATCH field means "leave it alone". Validate every
# supplied field before mutating either stored value so a bad
# partial update cannot produce a half-applied standing portrait.
standing = (
_validated_standing_portrait(
dependencies.assets, updates["standing_portrait"],
)
if standing_portrait_present else None
)
standing_expressions = (
_validated_standing_portrait_expressions(
dependencies.assets, updates["standing_portrait_expressions"],
)
if standing_expressions_present else None
)
except ValueError as exc:
return {"ok": False, "error": str(exc)}
if standing_portrait_present:
updates.pop("standing_portrait")
if standing is None:
cs.pop("standing_portrait", None)
else:
cs["standing_portrait"] = standing
if standing_expressions_present:
updates.pop("standing_portrait_expressions")
if standing_expressions is None:
cs.pop("standing_portrait_expressions", None)
else:
cs["standing_portrait_expressions"] = standing_expressions
if "voice_profile" in updates:
try:
voice = _validated_voice_profile(updates.pop("voice_profile"))
except ValueError as exc:
return {"ok": False, "error": str(exc)}
if voice is None:
cs.pop("voice_profile", None)
else:
cs["voice_profile"] = voice
explicit_hp_update = (
"hp" in updates
or "max_hp" in updates
or (
isinstance(updates.get("resources"), dict)
and isinstance(updates.get("resources", {}).get("hp"), dict)
)
)
# 消耗属性点
new_attrs = updates.get("attributes")
old_attrs = cs.get("attributes", {})
pool = cs.get("level_up_points", 0)
if isinstance(new_attrs, dict) and pool > 0 and isinstance(old_attrs, dict):
used = sum(max(0, int(new_attrs.get(k, 0)) - int(old_attrs.get(k, 0)))
for k in set(old_attrs) | set(new_attrs))
updates["level_up_points"] = max(0, pool - used)
cs.update(updates)
# 确保 attr_points_max 不低于当前属性总和，避免升级后编辑界面卡上线
try:
cur_attrs = cs.get("attributes", {})
if isinstance(cur_attrs, dict) and cur_attrs:
total = sum(int(v) for v in cur_attrs.values())
stored = cs.get("attr_points_max", 0)
if total > stored:
cs["attr_points_max"] = total
logger.info("attr_points_max 修正: %d -> %d (uid=%s)", stored, total, user_id)
except (TypeError, ValueError):
logger.warning("attr_points_max 自动修正失败: uid=%s", user_id, exc_info=True)
# 规范化技能格式
if "skills" in updates:
cs["skills"] = _normalize_skills(updates.get("skills", []), rule)
for source, row in zip((s for s in updates['skills'] if isinstance(s, (str, dict))), cs['skills']):
if isinstance(source, dict) and isinstance(source.get('id'), str) and source['id'] in accepted:
original = next(s for s in old_skills if isinstance(s, dict) and s.get('id') == source['id'])
row['id'] = original['id']
if 'improvable' in original:
row['improvable'] = original['improvable']
# 属性变化后可按规则补算 HP；若用户明确手填 HP，则尊重手填值。
new_attrs = updates.get("attributes")
if isinstance(new_attrs, dict) and not explicit_hp_update:
def formula_hp(attributes: dict) -> int:
return int(
rule.calculate_hp(attributes, cs.get("class", ""))
if rule
else calc_hp_from_rule(
attributes,
rules_dir=dependencies.rules.rules_dir,
language=getattr(inst, "language", ""),
)
)
try:
current_hp = int(cs.get("hp", 0) or 0)
if gm_authority:
# GM editor: rebuild max HP from the rule formula and keep the
# HP ratio. A downed character (HP <= 0) stays down.
lv_bonus = max(0, (cs.get("level", 1) - 1) * 5)
new_max = formula_hp(new_attrs) + lv_bonus
ratio = current_hp / max(1, int(cs.get("max_hp", 1) or 1))
new_current = max(1, round(new_max * ratio)) if current_hp > 0 else 0
else:
# Player level-up spend: apply only the formula delta the
# attribute change causes, so a GM-adjusted max HP survives
# and nothing is healed by rounding. A downed character
# (HP <= 0) stays down.
old_formula = (
formula_hp(old_attrs) if isinstance(old_attrs, dict) and old_attrs else None
)
delta = formula_hp(new_attrs) - old_formula if old_formula is not None else 0
new_max = max(0, int(cs.get("max_hp", 0) or 0) + delta)
new_current = max(1, current_hp + delta) if current_hp > 0 else 0
set_hp(cs, new_current, new_max)
logger.info("HP 重算: con=%s max_hp=%d hp=%d",
new_attrs.get("con", "?"), new_max, cs.get("hp", 0))
except Exception as exc:
logger.warning("属性变化后 HP 重算失败: %s", exc)
normalize_character_sheet(cs, rule)
inst.set_character_sheet(user_id, cs)
await dependencies.games.save_instance(inst)
try:
dependencies.save_character_card({
"character_name": inst.players[user_id].get("character_name", ""),
"character_sheet": cs,
})
except Exception:
logger.warning("角色卡同步入库失败: uid=%s", user_id, exc_info=True)
return {"ok": True}
def _validated_portrait(
dependencies: CharacterAssetDependencies,
portrait: Any,
) -> dict[str, str] | None:
return validate_character_portrait(dependencies, portrait)
def validate_character_portrait(
dependencies: CharacterAssetDependencies,
portrait: Any,
) -> dict[str, str] | None:
"""Validate and normalize a character portrait at the asset boundary."""
# Existing saves use an empty object as "no explicit portrait". Treat it
# exactly like null so editing unrelated profile fields remains possible.
if portrait is None or portrait == {}:
return None
if not isinstance(portrait, dict):
raise ValueError("头像数据无效")
kind = str(portrait.get("kind") or "")
if kind == "builtin":
portrait_id = str(portrait.get("id") or "")
if not portrait_id or len(portrait_id) > 100:
raise ValueError("内置头像编号无效")
return {"kind": "builtin", "id": portrait_id}
if kind == "upload":
asset_id = str(portrait.get("asset_id") or "")
if not asset_id or dependencies.avatar_file(asset_id) is None:
raise ValueError("上传头像不存在")
return {"kind": "upload", "asset_id": asset_id}
if kind == "generated":
asset_id = str(portrait.get("asset_id") or "")
if not asset_id or dependencies.generated_image_file(asset_id) is None:
raise ValueError("生成头像不存在")
return {"kind": "generated", "asset_id": asset_id}
raise ValueError("头像类型无效")
# ---- GAL P8：立绘 / 表情差分 / 音色绑定 --------------------------------------
_EXPRESSION_LABEL_RE = re.compile(r"^[a-z0-9][a-z0-9_-]{0,31}$")
MAX_STANDING_PORTRAIT_EXPRESSIONS = 16
_VOICE_PROFILE_ID_RE = re.compile(r"^personal:[a-f0-9]{32}$")
_PLUGIN_VOICE_PREFIX = "plugin:"
MAX_VOICE_PROFILE_ID_CHARS = 200
def _validated_standing_portrait(
dependencies: CharacterAssetDependencies, value: Any,
) -> dict[str, str] | None:
from src.webui.services.standing_portraits import validate_standing_portrait_ref
if dependencies.standing_portrait_file is None:
if value is None or value == {}:
return None
raise ValueError("立绘资产服务未配置")
return validate_standing_portrait_ref(value, dependencies.standing_portrait_file)
def _validated_standing_portrait_expressions(
dependencies: CharacterAssetDependencies, value: Any,
) -> dict[str, dict[str, str]] | None:
if value is None or value == {}:
return None
if not isinstance(value, dict):
raise ValueError("表情差分必须是对象")
if len(value) > MAX_STANDING_PORTRAIT_EXPRESSIONS:
raise ValueError(f"表情差分不能超过 {MAX_STANDING_PORTRAIT_EXPRESSIONS} 条")
normalized: dict[str, dict[str, str]] = {}
for label, reference in value.items():
if not _EXPRESSION_LABEL_RE.fullmatch(str(label or "")):
raise ValueError(f"表情标签无效：{label!r}（小写字母、数字、- 或 _，最长 32）")
item = _validated_standing_portrait(dependencies, reference)
if item is None:
raise ValueError(f"表情差分 {label!r} 缺少立绘引用")
normalized[str(label)] = item
return normalized
def _validated_voice_profile(value: Any) -> str | None:
"""Validate a voice id shape (speech vocabulary); existence is checked at use."""
if value is None or value == "":
return None
if not isinstance(value, str):
raise ValueError("音色绑定必须是字符串")
text = value.strip()
if not text or len(text) > MAX_VOICE_PROFILE_ID_CHARS:
raise ValueError("音色绑定无效")
if text.startswith("personal:") and not _VOICE_PROFILE_ID_RE.fullmatch(text):
raise ValueError("个人音色 ID 无效")
if text.startswith(_PLUGIN_VOICE_PREFIX) and len(text) <= len(_PLUGIN_VOICE_PREFIX):
raise ValueError("插件音色 ID 无效")
return text
def _project_npc_bindings(row: dict[str, Any], dependencies: CharacterDependencies) -> None:
"""Drop voice bindings whose profile no longer exists (fail closed).
失效引用保持未配置：投影省略后前端自然回退默认音色；未注入检查器时
原样带出，合成端仍会对已删除音色明确报错。
"""
exists = dependencies.voice_profile_exists
voice = str(row.get("voice_profile") or "")
if voice and voice.startswith("personal:") and exists is not None and not exists(voice):
row.pop("voice_profile", None)
async def update_npc_portrait(
dependencies: CharacterDependencies,
game_key: str,
npc_id: str,
portrait: Any,
) -> dict[str, Any]:
inst = dependencies.games.get_instance(
dependencies.games.parse_game_key(game_key),
)
if not inst:
return {"ok": False, "error": "游戏不存在"}
async with inst.authoritative_write() as write_entered:
if not write_entered:
return {
"ok": False, "error_code": "REWRITE_IN_PROGRESS",
"error": "GM 正在重写历史回合，请等待完成后重试",
}
return await _update_npc_portrait_authority(
dependencies, inst, npc_id, portrait,
)
def _resolve_npc_for_update(
dependencies: CharacterDependencies,
inst: GameInstance,
npc_id: str,
) -> tuple[str, dict[str, Any]] | None:
"""Locate an NPC (by key or id), materializing a lorebook entry if needed."""
npc_key = str(npc_id or "").strip()
if not npc_key:
return None
npc = inst.npcs.get(npc_key)
if npc is None:
for key, candidate in inst.npcs.items():
if str(candidate.get("id") or candidate.get("npc_id") or "") == npc_key:
npc_key, npc = key, candidate
break
if npc is None and dependencies.assets.lorebook and inst.world_id:
if dependencies.assets.content_projection is not None:
entries = dependencies.assets.content_projection.for_game(
inst, viewer_kind="gm", entry_type="npc",
)
entry = next(
(candidate for candidate in entries if str(candidate.get("id") or "") == npc_key),
None,
)
else:
entry = dependencies.assets.lorebook.get_entry(npc_key)
if entry and ("world_id" not in entry or entry.get("world_id") == inst.world_id) and entry.get("type") == "npc":
world_data = dependencies.assets.load_world_template(
inst.world_id,
str(getattr(inst, "language", "") or ""),
)
entry = localize_lorebook_entries([entry], world_data)[0]
name = str(entry.get("name") or npc_key)
npc = {
"name": name,
"character_name": name,
"tier": entry.get("tier", ""),
}
inst.npcs[npc_key] = npc
if npc is None:
return None
return npc_key, npc
async def _update_npc_portrait_authority(
dependencies: CharacterDependencies,
inst: GameInstance,
npc_id: str,
portrait: Any,
) -> dict[str, Any]:
resolved = _resolve_npc_for_update(dependencies, inst, npc_id)
if resolved is None:
return {"ok": False, "error": "NPC 不存在"}
_npc_key, npc = resolved
try:
normalized = _validated_portrait(dependencies.assets, portrait)
except ValueError as exc:
return {"ok": False, "error": str(exc)}
if normalized is None:
npc.pop("portrait", None)
else:
npc["portrait"] = normalized
await dependencies.games.save_instance(inst)
return {"ok": True, "portrait": npc.get("portrait")}
# Marks an NPC standing-portrait field the request did not carry.
UNCHANGED: Any = object()
async def update_npc_standing_portrait(
dependencies: CharacterDependencies,
game_key: str,
npc_id: str,
standing_portrait: Any = UNCHANGED,
expressions: Any = UNCHANGED,
) -> dict[str, Any]:
"""GAL P8：设置 NPC 的立绘主图与表情差分集（一次事务，全有或全无）。
与玩家角色卡同一语义：未提供的字段保持不变，显式 ``None`` 清除该字段。
"""
inst = dependencies.games.get_instance(
dependencies.games.parse_game_key(game_key),
)
if not inst:
return {"ok": False, "error": "游戏不存在"}
async with inst.authoritative_write() as write_entered:
if not write_entered:
return {
"ok": False, "error_code": "REWRITE_IN_PROGRESS",
"error": "GM 正在重写历史回合，请等待完成后重试",
}
resolved = _resolve_npc_for_update(dependencies, inst, npc_id)
if resolved is None:
return {"ok": False, "error": "NPC 不存在"}
_npc_key, npc = resolved
portrait_present = standing_portrait is not UNCHANGED
expressions_present = expressions is not UNCHANGED
if not portrait_present and not expressions_present:
return {"ok": False, "error": "未提供立绘字段"}
try:
# Validate every supplied field before mutating either one.
normalized = (
_validated_standing_portrait(dependencies.assets, standing_portrait)
if portrait_present else None
)
normalized_expressions = (
_validated_standing_portrait_expressions(dependencies.assets, expressions)
if expressions_present else None
)
except ValueError as exc:
return {"ok": False, "error": str(exc)}
if portrait_present:
if normalized is None:
npc.pop("standing_portrait", None)
else:
npc["standing_portrait"] = normalized
if expressions_present:
if normalized_expressions is None:
npc.pop("standing_portrait_expressions", None)
else:
npc["standing_portrait_expressions"] = normalized_expressions
await dependencies.games.save_instance(inst)
return {
"ok": True,
"standing_portrait": npc.get("standing_portrait"),
"standing_portrait_expressions": npc.get("standing_portrait_expressions"),
}
async def update_npc_voice(
dependencies: CharacterDependencies,
game_key: str,
npc_id: str,
voice_profile: Any,
) -> dict[str, Any]:
"""GAL P8：绑定 NPC 的固定音色（voice id，speech 同词汇）。"""
inst = dependencies.games.get_instance(
dependencies.games.parse_game_key(game_key),
)
if not inst:
return {"ok": False, "error": "游戏不存在"}
async with inst.authoritative_write() as write_entered:
if not write_entered:
return {
"ok": False, "error_code": "REWRITE_IN_PROGRESS",
"error": "GM 正在重写历史回合，请等待完成后重试",
}
resolved = _resolve_npc_for_update(dependencies, inst, npc_id)
if resolved is None:
return {"ok": False, "error": "NPC 不存在"}
_npc_key, npc = resolved
try:
normalized = _validated_voice_profile(voice_profile)
except ValueError as exc:
return {"ok": False, "error": str(exc)}
if normalized is None:
npc.pop("voice_profile", None)
else:
npc["voice_profile"] = normalized
await dependencies.games.save_instance(inst)
return {"ok": True, "voice_profile": npc.get("voice_profile")}
def grant_reward_to_sheet(sheet: dict[str, Any], reward: dict[str, Any]) -> None:
"""Write one settled reward onto a character sheet.
``item_ref`` 是**不透明身份 metadata**：本层不解释 D&D 规则，只保证它
随奖励一起落到背包行，让规则集侧能按身份取件（缺失时行为与旧版一致）。
"""
item_name = str(reward.get("name") or "").strip()
if not item_name:
return
grant_classified_item(
sheet,
item_name,
str(reward.get("category") or ""),
item_ref=str(reward.get("item_ref") or ""),
)
async def resolve_payment(
dependencies: CharacterDependencies,
game_key: str,
payment_id: str,
accepted: bool,
session_uid: str = "",
) -> dict[str, Any]:
inst = dependencies.games.get_instance(
dependencies.games.parse_game_key(game_key),
)
if inst is None:
return {"ok": False, "error": "游戏不存在"}
actor_uid = str(session_uid or "")
def grant_reward(sheet: dict[str, Any], reward: dict[str, Any]) -> None:
grant_reward_to_sheet(sheet, reward)
scene_image_payload: dict[str, Any] | None = None
async with inst.authoritative_write() as write_entered, inst._lock:
if not write_entered:
return {
"ok": False,
"code": "REWRITE_IN_PROGRESS",
"error": "GM 正在重写历史回合，请等待完成后再处理支付",
}
current = dependencies.games.get_instance(
dependencies.games.parse_game_key(game_key),
)
if current is None or id(current) != id(inst):
return {
"ok": False,
"code": "STALE_RUN",
"error": "对局已重开，请刷新后处理新的请求",
}
staged = type(inst).from_dict(deepcopy(inst.to_dict()))
# The persistence projection intentionally keeps only the recent log
# window. A transaction stage must retain the live aggregate's full
# in-memory history when it is committed back.
staged.log = deepcopy(inst.log)
result = resolve_proposal(
staged,
payment_id,
actor_uid=actor_uid,
accepted=bool(accepted),
grant_reward=grant_reward,
)
effect_group = result.get("effect_group")
if (
result.get("ok")
and isinstance(effect_group, dict)
and isinstance(effect_group.get("effects"), dict)
and dependencies.apply_economy_effects is not None
):
effect_payload = dict(effect_group.get("effects") or {})
memory_delta = dict(effect_payload.pop("memory_delta", {}) or {})
scene_image_prompt = str(
effect_payload.pop("scene_image_prompt", "") or ""
).strip()
if scene_image_prompt:
scene_image_payload = {
"scene_image_prompt": scene_image_prompt,
"state_update": deepcopy(effect_payload.get("state_update") or {}),
}
try:
# Deferred narrative effects are allowed to apply item
# candidates only under the transaction receipt produced by
# resolve_proposal. The callback shape remains compatible
# with legacy two-argument test/application adapters; the
# receipt travels on the staged aggregate and is never
# persisted.
staged._economy_authorized_receipt_id = str(
(result.get("transaction") or {}).get("id") or ""
)
await dependencies.apply_economy_effects(
staged, effect_payload,
)
except Exception:
logger.exception(
"经济提案关联效果提交失败，已回滚: game=%s proposal=%s",
game_key,
payment_id,
)
return {
"ok": False,
"code": "EFFECT_COMMIT_FAILED",
"error": "关联结果应用失败，本次结算未生效，请重试",
}
finally:
staged.__dict__.pop("_economy_authorized_receipt_id", None)
if memory_delta and dependencies.apply_economy_memory is not None:
queue_memory_delivery(
staged,
effect_group_id=str(effect_group.get("id") or ""),
memory_delta=memory_delta,
round_number=int(effect_group.get("round", progression_state.round_value(staged)) or 0),
)
result["effects_committed"] = complete_effect_group(
staged, str(effect_group.get("id") or ""),
)
if result["effects_committed"] and isinstance(result.get("outcome"), dict):
result["outcome"]["effects_status"] = "committed"
outcome = result.get("outcome")
if isinstance(outcome, dict):
_record_economy_outcome_in_round(staged, outcome)
proposal = result.get("proposal") or {}
if result.get("ok"):
uid = str(
proposal.get("payer_uid")
or proposal.get("recipient_uid")
or ""
)
amount = int(proposal.get("amount", 0) or 0)
name = staged.players.get(uid, {}).get("character_name", uid)
awaiting_party = bool(accepted and result.get("committed") is False)
record_health_event(
staged,
component="economy",
code=(
"economy_approved"
if awaiting_party
else "economy_committed"
if accepted
else "economy_declined"
),
severity="info",
title="经济提案已处理",
message=(
f"{name} 已确认，等待其他队员：{proposal.get('reason') or amount}"
if awaiting_party
else f"{name} {'确认' if accepted else '拒绝'}：{proposal.get('reason') or amount}"
),
)
if result.get("ok") or result.get("code") == "INSUFFICIENT_FUNDS":
_remove_stale_pending_notice(staged)
before_commit = type(inst).from_dict(deepcopy(inst.to_dict()))
before_commit.log = deepcopy(inst.log)
inst.replace_persisted_state_from(staged)
# Persist while still holding the aggregate lock. This closes the
# gap where restart could replace the run between mutation and save.
try:
await dependencies.games.save_instance(inst)
except Exception:
inst.replace_persisted_state_from(before_commit)
raise
if (
scene_image_payload is not None
and dependencies.schedule_economy_scene_image is not None
):
try:
result["scene_image_scheduled"] = bool(
dependencies.schedule_economy_scene_image(
inst, scene_image_payload,
)
)
except Exception:
logger.exception(
"经济结算后的场景图任务创建失败，权威结算已保留: game=%s",
game_key,
)
result["scene_image_scheduled"] = False
result["external_effects_committed"] = await _deliver_memory_outbox_locked(
dependencies, inst,
)
if not result.get("ok"):
return result
return {**result, "payment": proposal}
async def _deliver_memory_outbox_locked(
dependencies: CharacterDependencies,
instance: GameInstance,
) -> bool:
"""Deliver persisted memory effects; caller must hold the aggregate lock."""
reversals = pending_memory_reversals(instance)
pending = pending_memory_deliveries(instance)
if not reversals and not pending:
return True
reverse_memory = dependencies.reverse_economy_memory
changed = False
if reversals:
if reverse_memory is None:
return False
for delivery in reversals:
try:
reversed_ok = await reverse_memory(
instance.memory_namespace,
str(delivery.get("id") or ""),
)
except Exception:
logger.exception(
"经济结算记忆撤销失败，已保留待重试记录: game=%s delivery=%s",
instance.game_key,
delivery.get("id"),
)
return False
if not reversed_ok:
logger.error(
"经济结算记忆缺少可验证的撤销记录: game=%s delivery=%s",
instance.game_key,
delivery.get("id"),
)
return False
complete_memory_reversal(instance, str(delivery.get("id") or ""))
changed = True
apply_memory = dependencies.apply_economy_memory
if pending:
if apply_memory is None:
return False
for delivery in pending:
try:
await apply_memory(
instance.memory_namespace,
str(delivery.get("id") or ""),
dict(delivery.get("payload") or {}),
int(delivery.get("round", progression_state.round_value(instance)) or 0),
)
except Exception:
logger.exception(
"经济结算记忆 outbox 投递失败，已保留待重试记录: game=%s delivery=%s",
instance.game_key,
delivery.get("id"),
)
return False
complete_memory_delivery(instance, str(delivery.get("id") or ""))
changed = True
if changed:
# Memory delivery was the last outstanding piece of the settlement;
# the round's stale "settlement pending" notice can finally go.
_remove_stale_pending_notice(instance)
await dependencies.games.save_instance(instance)
return not (
pending_memory_deliveries(instance)
or pending_memory_reversals(instance)
)
async def drain_economy_outbox(
dependencies: CharacterDependencies,
instance: GameInstance,
) -> bool:
"""Retry durable external effects before allowing narrative progression."""
async with instance._lock:
current = dependencies.games.get_instance(instance.game_key)
if current is not instance:
return False
return await _deliver_memory_outbox_locked(dependencies, instance)
async def delete_character(
dependencies: CharacterDependencies,
game_key: str,
user_id: str,
*,
gm_authority: bool,
) -> dict[str, Any]:
"""Remove a seat. Once it has acted only the GM may (no delete-and-rejoin
reset); checked inside the authoritative write."""
inst = dependencies.games.get_instance(
dependencies.games.parse_game_key(game_key),
)
if not inst:
return {"ok": False, "error": "角色不存在"}
async with inst.authoritative_write() as write_entered:
if not write_entered:
return {
"ok": False, "error_code": "REWRITE_IN_PROGRESS",
"error": "GM 正在重写历史回合，请等待完成后重试",
}
player_side_after_acting = (
not gm_authority
and user_id in inst.players
and module_hooks.seat_has_acted(inst, user_id)
)
current = dependencies.games.get_instance(inst.game_key)
if current is not inst:
return {"ok": False, "code": "STALE_RUN", "error": "对局已重开，请刷新后重试"}
if player_side_after_acting:
return delete_requires_gm_failure()
return await _delete_character_authority(dependencies, inst, user_id)
async def _delete_character_authority(
dependencies: CharacterDependencies,
instance: GameInstance,
user_id: str,
) -> dict[str, Any]:
inst = instance
if not inst or user_id not in inst.players:
return {"ok": False, "error": "角色不存在"}
if len(inst.players) <= 1:
return {"ok": False, "error": "至少保留一个角色，无法删除最后一个"}
name = inst.players[user_id].get("character_name", user_id)
removed = await inst.remove_player(user_id)
if not removed:
return {"ok": False, "error": "角色不存在"}
cancel_proposals_for_player(inst, user_id)
inst.clear_private_messages(user_id)
await dependencies.games.save_instance(inst)
logger.info("角色已删除: %s (%s)", name, inst.game_key)
return {"ok": True}
async def create_player(
dependencies: CharacterDependencies, game_key: str, character: dict,
force_uid: str = "", assign_new_id: bool = False,
*, seat_token_uid: str = "", require_seat_proof: bool = False,
) -> dict[str, Any]:
inst = dependencies.games.get_instance(
dependencies.games.parse_game_key(game_key),
)
if not inst:
return {"ok": False, "error": "游戏不存在"}
inst = cast(GameInstance, inst)
async with inst.authoritative_write() as write_entered:
if not write_entered:
return {
"ok": False, "error_code": "REWRITE_IN_PROGRESS",
"error": "GM 正在重写历史回合，请等待完成后重试",
}
if dependencies.games.get_instance(inst.game_key) is not inst:
return {"ok": False, "code": "STALE_RUN", "error": "对局已重开，请刷新后重试"}
# Serialize membership changes with authoritative combat actions. A
# player joining during an active encounter must be enrolled in the
# ruleset's live turn state before the new seat is persisted.
state_lock = getattr(inst, "_lock")
async with state_lock:
return await _create_player_authority(
dependencies, inst, character, force_uid, assign_new_id,
seat_token_uid=seat_token_uid, require_seat_proof=require_seat_proof,
)
def _record_seat_claim(inst: GameInstance, uid: str) -> bool:
"""把"重新加入一个已有席位"记录为一次控制权认领。
新席位由 ``put_player`` 直接生成为 ``human``；只有当前是 ``ai`` /
``unclaimed`` 的已有席位才存在需要记录的转换，因此普通真人重连不会触发
``CONTROL_NOT_CLAIMABLE``。转换走 :func:`claim_seat`，与其它控制权变更是
同一个写入口。
返回是否真的发生了转换：认领是持久化状态，必须立刻落盘，否则真人认领过的
席位会在重启后变回 AI；而普通重连什么都没改，不能因此在热路径上多写一次盘。
"""
if is_human_controlled(inst, uid):
return False
claim_seat(inst, uid)
return True
def _seat_proof_missing(inst: GameInstance, uid: str, seat_token_uid: str, *, named: bool) -> bool:
"""Re-joining an existing seat needs that seat's token.
Naming another seat's uid is never proof. A session already bound to a
seat may rejoin it only while the seat has no credential yet (the
pre-token migration); once a credential exists only the token speaks.
"""
if seat_token_uid == uid:
return False
return named or room_access.has_seat_token(inst, uid)
_SEAT_PROOF_REQUIRED = {
"ok": False,
"error_code": "SEAT_TOKEN_REQUIRED",
"error": "加入已有席位需要该席位的链接",
}
async def _create_player_authority(dependencies: CharacterDependencies, inst: GameInstance, character: dict,
force_uid: str = "", assign_new_id: bool = False,
*, seat_token_uid: str = "", require_seat_proof: bool = False) -> dict[str, Any]:
requested_uid = str(character.get("user_id") or "").strip()
if (
require_seat_proof
and requested_uid in inst.players
and _seat_proof_missing(inst, requested_uid, seat_token_uid, named=True)
):
return dict(_SEAT_PROOF_REQUIRED)
if requested_uid and requested_uid in inst.players:
# 已有席位被重新加入：按控制权权威记录这次认领（已经是真人则是普通重连）。
if _record_seat_claim(inst, requested_uid):
await dependencies.games.save_instance(inst)
return {
"ok": True,
"user_id": requested_uid,
"character_name": inst.players[requested_uid].get("character_name", requested_uid),
"reused": True,
}
# uid 决策：GM 代建(assign_new_id)生成独立 uid；否则优先 force_uid（session 身份）或显式 requested_uid
if assign_new_id:
uid = "player_" + str(time.time_ns())[-12:]
elif force_uid:
if (
require_seat_proof
and force_uid in inst.players
and _seat_proof_missing(inst, force_uid, seat_token_uid, named=False)
):
return dict(_SEAT_PROOF_REQUIRED)
if force_uid in inst.players:
# 同上：session 身份命中的是已有席位，而不是新建席位。
if _record_seat_claim(inst, force_uid):
await dependencies.games.save_instance(inst)
return {"ok": True, "user_id": force_uid,
"character_name": inst.players[force_uid].get("character_name", force_uid),
"reused": True}
uid = force_uid
elif requested_uid:
uid = requested_uid
else:
uid = "player_" + str(time.time_ns())[-12:]
# A new seat is issued its token in the same write, so the slot must be
# writable before anything changes.
room_access.require_writable(inst)
max_players = max(1, int(room_access.max_players(inst) or 6))
if uid not in inst.players and len(inst.players) >= max_players:
return {
"ok": False,
"error": f"房间已满（最多 {max_players} 人）",
"error_code": "game_room_full",
}
rule = dependencies.rules.load_rule_for_game(inst)
rule_id = rule.rule_id if rule else "freeform_fantasy"
professional_character = False
runtime: Any | None = None
if rule is not None:
runtime = dependencies.rules.ruleset_registry.resolve(rule.template)
professional_character = runtime.capabilities.character_builder == "professional"
if professional_character:
try:
character = runtime.normalize_character_submission(
rule, character, getattr(inst, "language", ""),
)
except ValueError as exc:
error_code, error = professional_character_validation_error(exc)
return {
"ok": False,
"error_code": error_code,
"error": error,
}
canonical = character.get("ruleset_character")
binding = canonical.get("rule_binding") if isinstance(canonical, dict) else None
# 绑定由游戏在创建时决定；加入的角色只能匹配，不能反向建立绑定。
if not ruleset_runtime.binding(inst):
return {
"ok": False,
"error_code": "RULESET_BINDING_MISSING",
"error": "这个存档没有绑定专业规则版本（旧版存档），无法加入专业角色，请新建一局游戏。",
}
if not isinstance(binding, dict) or not inst.ruleset_binding_matches(binding):
return {
"ok": False,
"error_code": "INCOMPATIBLE_RULESET_CHARACTER",
"error": "角色与当前游戏的专业规则版本不兼容",
}
# 仅传了名字的轻量加入会自动生成默认角色卡。
has_full_sheet = bool(character.get("attributes") or character.get("equipment") or character.get("skills"))
if not has_full_sheet:
name = character.get("name") or character.get("character_name") or "冒险者"
try:
templates_base = (
dependencies.rules.rules_dir.parent
if dependencies.rules.rules_dir
else None
)
character = make_default_character(name, rule_id or "freeform_fantasy", templates_base, language=getattr(inst, "language", ""))
character["character_name"] = name
except Exception:
logger.exception("生成默认角色卡失败: %s", name)
attrs = character.get("attributes", {})
clamp_to_rule = _creation_bounds_enabled(rule)
if clamp_to_rule:
attrs = _clamp_attributes(rule, attrs)
rule_attrs = _get_rule_attrs_for_game(dependencies, inst)
total_points = sum(int(attrs.get(r["key"], 10)) for r in rule_attrs) if rule_attrs else 60
default_weapons = [{"name": "徒手", "type": "weapon", "damage": 2, "slot": "main_hand", "quality": "common"}]
hp = character.get("hp")
max_hp = character.get("max_hp")
# 声明了入场边界的规则（CoC 的 HP 由体质与体型派生）不接受客户端自报 HP。
if hp is None or max_hp is None or clamp_to_rule:
hp = (
rule.calculate_hp(attrs, character.get("class", ""))
if rule
else calc_hp_from_rule(
attrs,
rule_id,
dependencies.rules.rules_dir,
character.get("class", ""),
language=getattr(inst, "language", ""),
)
)
max_hp = hp
default_class = rule.classes[0]["name"] if (rule and rule.classes) else "冒险者"
if len(str(character.get("background", ""))) > MAX_BIO_CHARS:
return {"ok": False, "error": f"角色背景过长（上限 {MAX_BIO_CHARS} 字）"}
damage_dice_error = _validate_damage_dice(character.get("equipment"))
if damage_dice_error:
return {"ok": False, "error": damage_dice_error}
starter_equip, starter_inv = build_starter_items(rule, character.get("class") or default_class)
cs = {
"race": character.get("race", "人类"),
"class": character.get("class") or default_class,
"identity": character.get("identity", {}),
"level": 1, "xp": 0,
"attributes": attrs,
"hp": hp, "max_hp": max_hp,
"equipment": character.get("equipment") or starter_equip or default_weapons,
"inventory": character.get("inventory") or starter_inv,
"key_items": character.get("key_items", []),
"skills": _normalize_skills(character.get("skills", []), rule, attributes=attrs, clamp=clamp_to_rule),
"background": character.get("background", ""),
"deceased": False,
"gold": character.get("gold", 30),
"currency": character.get("currency", {}),
"portrait": character.get("portrait", {}),
"attr_points_max": total_points,
}
if professional_character:
cs["rule_binding"] = deepcopy(character["rule_binding"])
cs["ruleset_character"] = deepcopy(character["ruleset_character"])
cs["armor_class"] = int(character.get("armor_class", 10) or 10)
# 初始化 special_stats（理智值/幸运值/内力等）
try:
if rule:
for ss in rule.special_stats:
max_val = ss.get("max", 99)
init_val = initial_special_stat_value(ss, attrs)
cs[ss["key"]] = init_val
cs[f"max_{ss['key']}"] = max_val
except Exception as exc:
logger.exception("初始化特殊属性失败: %s", exc)
normalize_character_sheet(cs, rule)
# 规则显式声明了服务端建卡边界（CoC 7e 的 character_creation.clamp_to_rule_bounds）时，
# 属性 clamp 只是第一道；技能点预算、技能条数、单技能上限也必须在这里按同一套规则
# 校验，否则绕过前端直接调 API 就能把超预算角色带进局内。
if clamp_to_rule:
rejection = creation_bounds_violation(rule, cs)
if rejection is not None:
return rejection
player = {
"character_name": character.get("character_name") or character.get("name") or "冒险者",
"character_sheet": cs,
}
inst.put_player(uid, player)
# 新建席位由 put_player/ensure_control 直接生成为 human 控制，无需认领转换。
if isinstance(runtime, PlayerJoinRuntime):
try:
runtime.on_player_join(inst, uid)
except Exception:
# Do not leave a seat visible without a corresponding ruleset
# combat actor if the optional live-state hook rejects the join.
inst.players.pop(uid, None)
raise
# The new seat's share credential; the plaintext leaves only in this
# response. A GM-created seat's token lets the GM hand out its link.
# The GM seat itself never holds one (e.g. the GM recreating its character).
seat_token = "" if uid == str(inst.gm_uid or "") else room_access.issue_seat_token(inst, uid)
dependencies.save_character_card({
**player,
"rule_id": rule_id,
"rule_name": rule.rule_name if rule else rule_id,
"rule_version": str(rule.template.get("rule_version") or "") if rule else "",
"mechanics": rule.mechanics if rule else "",
"language": getattr(inst, "language", ""),
})
await dependencies.games.save_instance(inst)
created: dict[str, Any] = {"ok": True, "user_id": uid, "character_name": player["character_name"]}
if seat_token:
created["seat_token"] = seat_token
return created