Files
2026-10-21 15:20:52 +08:00

536 lines
31 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""技能关联计算与维护说明
技能关联由 `core/basic/skill/effects.py` 的 `SkillEffectEngine` 统一处理。
`lv`、VP 等输入变化不再通过历史倍率乘除更新目标,而是重新解析最终状态并汇总本次关联贡献。
## 技能身份与引用维护
职业模块导入自己的 Ref;技能类声明 `ref = Ref.XXX`,`code` 从 ref 派生,name 只负责显示。
角色初始化根据技能数据 prefix 校验职业范围,并创建独立的 ref/code 索引;输出圣骑士等职业变体复用数据来源范围。
SkillEffect.skills、except_skills、Read.skill、level_source 统一接受 Ref,旧名称字符串继续兼容。
公共技能引用只匹配当前角色实际绑定的技能实例。GetSkill(Ref.XXX) 查询引用,GetSkillByCode 查询原始 code;
字符串不会隐式当作 code。重复绑定、跨职业引用直接报错;新增规则应使用 Ref,旧失配名称继续保留 SKIP 诊断。
引用生成、公共分组、编辑器中文提示的维护说明见 core/character/refs/base.py。
SkillRef.name 只影响提示,不参与相等性和哈希;技能名称变更不会改变引用身份。
## 计算阶段
1. `SetSkills` 批量读取基础等级、VP、UP。传入的等级是基础等级,不含关联技能提供的等级;重复传入同一配置不会重复加级。
2. 解析等级和参数依赖。`lv`、`ratio`、`maxLayer` 是状态属性;关联到这些属性的规则以及 `level_source` 组成依赖图,按拓扑顺序求值。关联规则读取其他技能等级时,依赖会自动记录。循环、缺失共享等级源、计算过程中改变状态依赖会报出具体技能。
3. 基于最终状态收集伤害、冷却和角色属性贡献,并在装备提供的基础值之上应用一次。
`calc` 在处理配置、装备、时装、徽章等阶段使用同一个批次。需要读取等级时,可以先解析状态;伤害及角色属性关联在准备阶段结束后统一生效。
## 后续在哪里维护
| 新增或修改的影响 | 维护位置 | 要点 |
| --- | --- | --- |
| 技伤、冷却、攻击段数等普通关联 | 对应 `core/character/GF/职业/角色.py` 的技能类 `associate` | 通常不改引擎;目标属性应有默认值 |
| 随 VP、等级或自身参数变化的关联 | `associate` 中的 `Read`、`LevelTable`、`Select`、`when` | 动态目标使用 `effect`,数值和条件使用表达式 |
| 技能给其他技能加级 | `SkillEffect(attribute=SkillAttribute.LEVEL, operation=Operation.ADD, ...)` | value 直接填整数增量;只给已学习的目标加级 |
| 生命之树这类共享等级 | 目标技能的 `level_source` | 指向源技能名称,不再写等级 getter/no-op setter |
| 新增会被关联修改、并影响后续关联的参数 | 技能类默认值 + `effect_types.py` 的 `SkillAttribute`、`STATE_ATTRIBUTES` | 新参数同时检查读写范围;状态属性自动属于输入属性 |
| 新增外部配置输入,改变它需要重新计算关联 | 技能类、配置赋值入口 + `INPUT_ATTRIBUTES` | 例如现有 `vp`;单纯登记不提供跨技能参数读取依赖 |
| 新增职业专用输出属性 | 职业文件内继承 `ProfessionAttribute` 定义枚举,并在目标技能设置数值默认值 | 成员紧跟中文文档字符串,IDE 可显示含义;不可通过 `Read` 读取 |
| 新增角色属性关联 | `effect_types.py` 的 `CharacterAttribute`,并核对 `character/base.py` 的 `SetStatus` | 当前仅支持对应标量字段;特殊运算需单独实现 |
| 新增运算符、目标选择方式、跨技能非等级依赖 | `base.py` 的公共关联入口 + `effects.py` | 属于公共能力扩展,需配套单元测试 |
| 装备改变技能等级或倍率 | 原装备规则及 `character/base.py` 的公共修改方法 | 使用已有批处理方法;低层直接写入须在 `batch()` 内 |
| 仅影响单次技能展示/输出的模式、VP、UP 变换 | 原技能 `setMode`、`vp_N`、输出计算方法 | `skillInfo()` 使用临时副本,不要误改为持久关联 |
普通职业规则优先放在职业文件。只有新增公共能力时才修改引擎;不要再用属性 setter 手动撤销旧倍率并乘入新倍率。
## 声明示例和数值约定
关联统一使用 `effect_types.py` 的不可变声明,`associate` 继续是数组,目标仍然使用技能名称数组:
```python
associate = [
SkillEffect(skills=['目标技能名称'], value=1 + Percent(LevelTable(data0))),
SkillEffect(attribute=SkillAttribute.LEVEL, operation=Operation.ADD,
skills=['目标精通名称'], value=Read(SkillAttribute.LEVEL) + 1),
SkillEffect(attribute=Power(13), operation=Operation.ADD,
skills=['破军升龙击'], value=Percent(LevelTable(data2))),
]
```
`SkillEffect` 作用于技能,`CharacterEffect` 作用于角色。Operation 明确区分加算、乘算、覆盖;
新声明的 value 是实际贡献值,不再自动按百分比缩放,也不根据属性名称推测冷却公式。
例如增加 20% 技伤使用 `1 + Percent(20)`,缩减 10% 冷却使用 `1 - Percent(10)`。
旧 data[0] 不为零的规则迁移时必须保留零级基准的除法/减法,不能直接省略。
`Read(SkillAttribute.RATIO, '技能名称')` 可分析跨技能状态依赖;省略名称表示读取自身。
`LevelTable` 按来源最终等级取表,`Select` 按条件选值,`when=Equals(Read(SkillAttribute.VP), 1)` 控制是否生效。
条件和未选分支也声明全部依赖,建图阶段不计算实际数值,避免在临时状态下发生除零或遗漏依赖。
当前 Read 支持等级、ratio、maxLayer、VP、UP,不能读取尚未汇总的伤害/冷却输出。
武器类型用 `weapon=['光剑']`,武器分类用 `WeaponCategoryContains('传世武器')`;武器信息来自提前完成的 SetDetail。
编号属性通过 `Power(n)`、`Hit(n)`、`ExtraData(n)`、`ExtraPower(n)` 构造,分别对应 powerN、hitN、dataplusN、powerplusN。
名称数组和等级表在构造时复制为元组,外部修改原数组不会影响已注册声明。
新类型、枚举成员、字段和构造函数的中文文档可供编辑器悬停和补全查看。
GF 职业的关联现已使用类型化声明,原 associate、动态 effect 和专用更新函数紧邻新实现注释保留。
男/女大枪的等级目标、黑暗武士的连招目标仍在 effect 内动态组装技能名称数组;贡献值和条件使用类型化声明。
模式、VP、UP 对临时技能副本的变换保留原实现,不属于持久关联。
旧名称失配显式使用 `MissingTargetPolicy.SKIP`,继续记录 unresolved_targets;新规则默认报错,不要为了消除报错批量设置 SKIP。
原先未生效的错误角色字段及特殊字典键保持原有效计算行为,迁移不同时修正规则。
### 职业专用属性
公共 `SkillAttribute` 仅维护通用字段。专用输出字段就近放在职业模块中,例如:
```python
class HunterAttribute(ProfessionAttribute):
'''猎人专用输出属性。'''
MARK_RATIO = 'markRation'
'''每层标记对应的技能伤害增量。'''
# 目标技能先定义 markRation = 0,然后声明:
associate = [SkillEffect(attribute=HunterAttribute.MARK_RATIO,
operation=Operation.ADD, value=Percent(LevelTable(data0)))]
```
不同职业可各自定义同名字段。枚举值必须对应目标技能的实际数值属性;不能重定义公共字段或编号分段字段。
专用枚举只用于最终输出,不自动加入状态依赖。将来某参数需要被后续关联读取时,应先登记公共状态/读写范围,
再使用 Read 表达式,不能直接读取尚未汇总的输出。条件可组合 `All`、`Not`、`GreaterThan`;
角色输出类型用 `OutputTypeStartsWith` 判断,所有条件均在建图时提供完整依赖。
## 职业规则
`associate` 仅接受 SkillEffect/CharacterEffect,旧字典会直接报错。动态 `effect(old, new)` 作为目标构造入口,在收集模式下以 `(0, 最终等级)` 执行;公共入口记录贡献,不修改目标。每次收集都从原始 `associate` 声明开始,取消 VP 时不会保留上一次分支。
动态规则只能生成关联声明,不能在收集过程中直接修改角色或目标技能。需要修改关联产生方式时,应改用声明,而不是另写参数 setter 撤销旧倍率。元素师原先的 `ratio` 补偿 setter 已移除,避免与公共引擎重复应用。
引擎在创建时保存初始声明和状态默认值。动态分支应新建并赋值 `self.associate`,不要对共享的类级列表执行 `append` 或原地改写;也不要把运行后修改 `associate` 当作注册新规则。一次重算可能多次调用 `effect`,因此不能在其中累计计数、直接调用角色加成方法或依赖随机结果。
依赖图目前自动识别三类依赖:共享等级源、收集时跨技能 `.lv` 读取、指向状态属性的关联。
类型化表达式的 Read 额外提供跨技能状态参数依赖;普通 Python 代码直接读取其他技能的 `.ratio` 仍不会自动建立依赖。
新规则请使用 Read;单独加入 INPUT_ATTRIBUTES 只会触发重算,不能解决求值先后问题。
本阶段仍使用技能级拓扑图,因此同一技能的状态回写、相互引用等可能被保守判定为循环;尚未升级为属性级依赖图。
初次建图后新出现的等级读取或状态目标会报错;不要靠最终状态临时增加依赖边。
每条贡献包含来源技能、条目索引、目标、属性、运算及数值,可从 `character.skill_effects.contributions` 检查。角色属性支持原 `SetStatus` 中相应的标量加算/乘算字段。
同一属性的汇总顺序为:基础值(或覆盖值)→ 加算贡献 → 乘算贡献。不同来源给出不同覆盖值会报冲突;不会依赖遍历顺序选择最后一个。赋零可以直接从基础值重建,不需要除以旧倍率。
## 共享等级
例如生命之树:
```python
level_source = '创世'
```
生命之树跟随创世最终等级,包含创世获得的装备及技能等级加成。目标自身的 `maxLv = 1` 不会截断共享等级。目标不接受独立等级或独立加级;需要提升等级时应提升源技能。
共享等级技能默认不独立提供关联效果,保留原 getter/no-op setter 的行为。如职业规则确实要求共享等级技能独立提供关联,可显式设置 `association_source = True`。
现有 11 个共享等级技能已迁移为 `level_source` 声明。
## 装备和外部修改
`SetStatus`、`AddSkillLv`、`SetSkillRation`、`SetSkillCD`、`SetSkillCDRecover` 已自动批处理。完整 `calc` 中的装备代码也在批次内,因此原有装备的直接属性写入会保留为关联计算的基础值。
新增低层直接修改必须放进批次,不能修改已叠加关联贡献的结果后再要求引擎推断基础值:
```python
with character.skill_effects.batch():
skill.skillDamage *= 1.15
```
`resolve()` 在没有变化时不重复应用;批次重建会先恢复原来的基础值。`skillInfo()` 的临时拷贝不属于引擎管理对象,其模式、VP、UP 变换仍在临时拷贝上执行。
这不表示装备自身也已经与顺序无关:装备仍按原流程执行,例如先由 `calc_suits()` 生成套装积分,再让装备读取积分。当前批次解决的是装备修改与技能关联贡献之间的基线问题,没有为装备建立逐来源贡献账本和依赖图,也不保证对同一实例反复执行整套装备流程不会累计。
定位引擎内部问题时,`_levels` 保存基础及外部等级,`_state_base` 保存状态参数基线,`_output_base` 保存应用技能输出贡献之前的属性值。它们是内部状态,不应由职业或装备规则直接修改;外部输入走公共修改方法或批次入口。
## 诊断和验证
找不到目标的旧关联仍按原逻辑跳过,同时记录在 `unresolved_targets`,供职业规则审查。例如刺客现存的“终结追击”名称失配;本次未推测其数值意图并修改规则。
类型化声明的缺失目标/引用、错误属性/运算、非有限数值会报错,避免新规则继续静默失效。
测试覆盖女弹药 2.32 场景及后续升降级、旅人条件、大枪加级、VP 切换、装备基线、角色属性、共享等级及链式共享、零覆盖、赋值冲突和循环依赖。全职业测试使用全部学习等级,在无 VP/VP1/VP2 下交换输入顺序,并另测搭配职业武器的情况。
重构前后 72 个模块的默认满技能配置对比:伤害/冷却的实质变化集中在女弹药、旅人、男女大枪已确认的问题;元素师统一了 `skillRation`/`skillDamage` 的分配,总伤害保持一致。该验证针对本地计算规则,不等同于游戏实测或线上验证。
新增规则时,在 `tests/unit/test_skill_effects.py` 补充对应场景,重点验证正反设置顺序、等级升降或禁用、条件开关、外部加成后再次重建。公共依赖机制变更还需验证循环及冲突报错。
从仓库根目录执行:
```sh
backend/.venv/bin/pytest backend/tests/unit/test_skill_effects.py -q
backend/.venv/bin/pytest backend/tests/unit/test_typed_skill_effects.py -q
backend/.venv/bin/pytest backend/tests/smoke/test_calc_availability.py -k configuration_order -q
```
第二项验证技能配置顺序(包括搭配武器的场景),不验证装备执行顺序。修改公共引擎后再运行完整 `backend/tests` 回归。
"""
from core.character.refs.base import resolve_skill
from collections import defaultdict
from contextlib import contextmanager
from dataclasses import dataclass
from functools import wraps
from graphlib import CycleError, TopologicalSorter
from math import prod
from .effect_types import (
CharacterAttribute, CharacterEffect, EffectDeclaration, MissingTargetPolicy, Operation,
SkillEffect, STATE_ATTRIBUTES,
)
# 状态属性定义在 effect_types;技能类仍须提供相应默认值。
# 外部赋值触发重算的输入;登记在此不等于自动跟踪跨技能的该属性读取。
INPUT_ATTRIBUTES = STATE_ATTRIBUTES | {'vp', 'up', 'comboCol', 'stage', 'currentMode'}
# CharacterBase.SetStatus 实际支持的标量属性。
# 扩展前核对 SetStatus 的实际运算语义;特殊字段不能仅加入白名单。
CHARACTER_ATTRIBUTES = frozenset(attribute.value for attribute in CharacterAttribute)
_MISSING = object()
@dataclass(frozen=True)
class SkillContribution:
"""类型化效果求值后生成的不可变贡献,记录来源、目标及实际运算值。"""
source: object
"""提供本条效果的来源技能实例。"""
index: int
"""同一来源收集到的贡献序号,用于稳定汇总顺序。"""
target: object
"""接受贡献的技能或角色实例。"""
attribute: str
"""目标实例的底层属性名称。"""
operation: str
"""汇总运算:'+' 加算、'*' 乘算、'=' 覆盖。"""
value: float
"""已经完成等级取值、百分比转换和条件求值的实际贡献值。"""
def skill_effect_batch(method):
"""将角色修改方法包入关联批次,使外部加成写入基础值,并在批次结束后重算。"""
@wraps(method)
def wrapped(self, *args, **kwargs):
"""有引擎时在批次内调用原方法;角色尚未初始化引擎时直接调用。"""
engine = getattr(self, 'skill_effects', None)
if engine is None:
return method(self, *args, **kwargs)
with engine.batch():
return method(self, *args, **kwargs)
return wrapped
class SkillEffectEngine:
"""技能关联引擎:管理输入基线,按依赖顺序解析状态,再汇总技能和角色输出。
每个角色独立持有一个实例;仅接受类型化声明,临时技能副本不受其管理。"""
def __init__(self, character):
"""绑定角色的技能实例,校验声明,并保存等级、状态参数和关联的初始基线。"""
self.character = character
self.skills = tuple(sorted(character.skills, key=lambda skill: skill.id))
self._members = {id(skill) for skill in self.skills}
self._levels = {skill: skill._lv for skill in self.skills}
self._state_base = {
(skill, attr): getattr(skill, attr)
for skill in self.skills for attr in STATE_ATTRIBUTES - {'lv'}
if hasattr(skill, attr)
}
self._associations = {skill: skill.associate for skill in self.skills}
for skill, declarations in self._associations.items():
for declaration in declarations:
self._validate_declaration(skill, declaration)
self._output_base = {}
self._depth = 0
self.resolving = False
self.collecting = None
self._discovering = False
self._pending = []
self._reads = set()
self._dirty = True
self._state_dirty = True
self.contributions = ()
self.unresolved_targets = set()
def owns(self, skill):
# skillInfo 使用浅拷贝,仅用于本次技能输出计算。
"""判断技能是否为本引擎管理的原实例,排除 skillInfo 创建的临时副本。"""
return id(skill) in self._members
def changed(self):
"""标记状态与输出需要重算;不在批次内时立即解析。"""
self._dirty = self._state_dirty = True
if not self._depth:
self.resolve()
def set_level(self, skill, value):
"""按当前最终等级与 value 的差值调整基础等级,避免装备加级重复计入关联加级。"""
if skill.level_source is not None:
return
# 装备 += 加级只调整输入,不能把已计算的精通加级写回基础等级。
# 配置中的绝对基础等级通过 set_input_level 写入。
self._levels[skill] = min(skill.maxLv, max(0, self._levels[skill] + int(value) - skill.lv))
self.changed()
def set_input_level(self, skill, value):
"""将 value 作为配置中的基础等级写入并标记待重算,不立即解析;共享等级技能忽略此输入。"""
if skill.level_source is not None:
return
self._levels[skill] = min(skill.maxLv, max(0, int(value)))
self._dirty = self._state_dirty = True
def set_attribute(self, skill, name, value):
"""写入外部输入属性;状态属性同时更新基线,并触发或排队重算。"""
if name in STATE_ATTRIBUTES:
self._state_base[skill, name] = value
object.__setattr__(skill, name, value)
self.changed()
def is_learned(self, skill):
"""按基础等级判断是否已学习;共享等级技能按其最终等级判断。"""
if skill.level_source is not None:
return skill.lv > 0
return self._levels[skill] > 0
def read_level(self, skill):
"""收集时记录跨技能等级依赖;普通读取时按需解析尚未更新的状态。"""
if self.collecting is not None and self.collecting is not skill:
self._reads.add(skill)
elif self._state_dirty and not self.resolving:
self.resolve_states()
def record(self, source, target, attribute, operation, value):
"""将已求值的效果追加到当前来源的待汇总贡献中,不直接修改目标。"""
self._pending.append(SkillContribution(
source, len(self._pending), target, attribute, operation, value,
))
def missing_target(self, source, name):
"""记录被跳过的目标引用及来源技能,供职业规则审查。"""
self.unresolved_targets.add((source.name, name))
@staticmethod
def _validate_declaration(source, declaration):
"""校验来源技能的单条声明类型;旧字典及其他不支持的类型直接报错。"""
if not isinstance(declaration, (SkillEffect, CharacterEffect)):
raise TypeError(
f'技能 {source.name} 的 associate 仅接受 SkillEffect/CharacterEffect,'
f'不支持 {type(declaration).__name__}'
)
def collect_effect(self, source, declaration: EffectDeclaration):
"""将来源技能的一条类型化声明解析为目标贡献。
建图阶段仅收集依赖和可能的状态目标,不计算表达式或按临时条件裁剪依赖。
求值阶段检查前置条件、武器和 when 条件,再记录实际贡献;必须在来源收集上下文中调用。"""
self._validate_declaration(source, declaration)
if self.collecting is not source:
raise RuntimeError('类型化效果只能由技能关联引擎收集')
for dependency in declaration.dependencies():
target = dependency.target(source)
if target is not source and dependency.attribute.value in STATE_ATTRIBUTES:
self._reads.add(target)
attribute = declaration.attribute.value
if isinstance(declaration, CharacterEffect):
targets = [self.character]
else:
targets = self.character.GetSkills('all', True) if declaration.skills == '*' else []
if declaration.skills != '*':
for name in declaration.skills:
if name in declaration.except_skills:
continue
target = resolve_skill(self.character, name)
if target is None:
if declaration.missing_target == MissingTargetPolicy.ERROR:
raise ValueError(f'类型化效果目标不存在: {source.name} -> {name}')
self.missing_target(source, name)
else:
targets.append(target)
excluded = {id(target) for ref in declaration.except_skills
if (target := resolve_skill(self.character, ref)) is not None}
targets = [target for target in targets if id(target) not in excluded]
for target in targets:
if attribute in STATE_ATTRIBUTES - {'lv'} and (target, attribute) not in self._state_base:
raise ValueError(f'类型化效果状态缺少默认值: {source.name} -> {target.name}.{attribute}')
if self._discovering:
# 保留 precondition 中的等级读取依赖,但不根据临时结果裁剪依赖边。
# 新条件优先通过 when 声明。
source.precondition()
value = 0
else:
if not source.precondition():
return
if declaration.weapon is not None and self.character.GetWeaponType()[0] not in declaration.weapon:
return
if declaration.when is not None and not declaration.when.evaluate(source):
return
value = declaration.evaluate(source)
for target in targets:
if target is not self.character and attribute == 'lv':
if target.level_source is not None:
continue
if not self._discovering and declaration.operation == Operation.ADD and not self.is_learned(target):
continue
self.record(source, target, attribute, declaration.operation.value, value)
def _collect(self, skill, *, discover=False):
# 一个重算周期会多次调用 effect;职业实现只能构造声明并调用公共收集入口。
"""收集单个技能的贡献与读取依赖,并在结束时清理收集上下文。
discover=True 表示仅发现依赖;每次从初始声明开始,避免残留上次动态分支。"""
self._pending = []
self._reads = set()
if not skill.association_source:
return (), set()
# 动态 VP 分支可能仅在启用时覆盖声明;每次恢复初始声明,
# 确保取消 VP 后不会保留上次的分支。
skill.associate = self._associations[skill]
self.collecting = skill
self._discovering = discover
try:
if skill.lv > 0:
skill.effect(0, skill.lv)
elif discover:
# 等级覆盖可能激活尚未学习的来源,因此建图时也要收集其静态依赖。
for declaration in skill.associate:
self.collect_effect(skill, declaration)
return tuple(self._pending), self._reads.copy()
finally:
self.collecting = None
self._discovering = False
@staticmethod
def _combine(base, contributions):
"""按覆盖、加算、乘算的顺序汇总贡献;不同来源给出不同覆盖值时抛出冲突。"""
contributions = sorted(contributions, key=lambda item: (item.source.id, item.index))
overrides = {item.value for item in contributions if item.operation == '='}
if len(overrides) > 1:
sources = ', '.join(item.source.name for item in contributions if item.operation == '=')
raise ValueError(f'关联赋值冲突: {sources}')
if overrides:
base = overrides.pop()
return (base + sum(item.value for item in contributions if item.operation == '+')) * prod(
item.value for item in contributions if item.operation == '*'
)
def _restore_outputs(self):
"""撤销上次技能输出贡献并恢复基础值;原先不存在的属性一并删除。"""
for (target, attr), value in self._output_base.items():
if value is _MISSING:
object.__delattr__(target, attr)
else:
object.__setattr__(target, attr, value)
self._output_base.clear()
def resolve_states(self):
# 依赖来源:level_source、跨技能 lv 读取、指向 STATE_ATTRIBUTES 的关联。
# 类型化 Read 可声明跨技能状态参数依赖;普通 Python 属性读取仍不会自动建边。
"""恢复输入基线、建立依赖图,并按拓扑顺序解析最终等级和状态参数。
共享等级源缺失、循环依赖或求值时新增依赖与状态目标均报错;本阶段不应用输出贡献。"""
if self.resolving or not self._state_dirty:
return
self.resolving = True
try:
self._restore_outputs()
self.unresolved_targets.clear()
for skill, lv in self._levels.items():
skill._lv = min(skill.maxLv, max(0, lv))
for (skill, attr), value in self._state_base.items():
object.__setattr__(skill, attr, value)
graph = {skill: set() for skill in self.skills}
for skill in self.skills:
if skill.level_source is not None:
source = resolve_skill(self.character, skill.level_source)
if source is None:
raise ValueError(f'共享等级源技能不存在: {skill.name} -> {skill.level_source}')
graph[skill].add(source)
contributions, reads = self._collect(skill, discover=True)
graph[skill].update(reads)
for item in contributions:
if item.target is not self.character and item.attribute in STATE_ATTRIBUTES:
graph[item.target].add(skill)
try:
order = tuple(TopologicalSorter(graph).static_order())
except CycleError as exc:
names = ' -> '.join(skill.name for skill in exc.args[1])
raise ValueError(f'技能关联状态存在循环依赖: {names}') from exc
states = defaultdict(list)
for skill in order:
# 拓扑顺序保证上游等级和参数贡献已完成计算。
if skill.level_source is not None:
skill._lv = resolve_skill(self.character, skill.level_source).lv
contributions, reads = self._collect(skill)
if not reads.issubset(graph[skill]):
raise ValueError(f'技能关联状态依赖随计算改变: {skill.name}')
for item in contributions:
if item.target is self.character or item.attribute not in STATE_ATTRIBUTES:
continue
if skill not in graph[item.target]:
raise ValueError(f'技能关联状态目标随计算改变: {skill.name} -> {item.target.name}')
key = item.target, item.attribute
states[key].append(item)
if item.attribute == 'lv':
base = self._levels[item.target]
value = self._combine(base, states[key])
item.target._lv = min(item.target.maxLv, max(0, int(value)))
else:
base = self._state_base[key]
object.__setattr__(item.target, item.attribute, self._combine(base, states[key]))
self._state_dirty = False
finally:
self.resolving = False
def resolve(self):
"""先解析最终状态,再按目标属性汇总并应用输出贡献;没有待处理变化时直接返回。"""
if self.resolving or not self._dirty:
return
self.resolve_states()
self.resolving = True
try:
self._restore_outputs()
contributions = []
grouped = defaultdict(list)
for skill in self.skills:
items, _ = self._collect(skill)
contributions.extend(items)
for item in items:
if item.target is not self.character and item.attribute in STATE_ATTRIBUTES:
continue
grouped[item.target, item.attribute].append(item)
for (target, attr), items in grouped.items():
base = getattr(target, attr, _MISSING)
self._output_base[target, attr] = base
if base is _MISSING and not any(item.operation == '=' for item in items):
raise AttributeError(f'技能关联属性不存在: {items[0].source.name} -> {attr}')
object.__setattr__(target, attr, self._combine(0 if base is _MISSING else base, items))
self.contributions = tuple(contributions)
self._dirty = False
finally:
self.resolving = False
@contextmanager
def batch(self):
# 最外层先撤销技能输出贡献,让装备等外部写入落在基础值上,再统一重建。
# 这里只隔离技能贡献,不解决装备规则彼此之间的执行顺序依赖。
"""批量接收外部修改,在最外层批次正常结束后统一重算。
进入最外层时先撤销技能输出贡献,使修改落在基础值上;支持嵌套,但不处理装备间的顺序依赖。"""
outermost = not self._depth
if outermost:
self.resolving = True
try:
self._restore_outputs()
finally:
self.resolving = False
self._dirty = self._state_dirty = True
self._depth += 1
completed = False
try:
yield
completed = True
finally:
self._depth -= 1
if completed and not self._depth:
self.resolve()