feat:优化职业技能关联实现

This commit is contained in:
2026-10-13 15:57:28 +08:00
parent 15c3bff467
commit 7dc5f4cde5
20 changed files with 875 additions and 161 deletions
+7 -1
View File
@@ -1,9 +1,9 @@
from collections.abc import Mapping
import sys
from typing import Any, Literal
from core.abstract.character import CharacterProperty
from core.basic.skill import Skill
from core.basic.skill_effects import SkillEffectEngine, skill_effect_batch
from core.basic.equipment import EquEffect, OathSuit, OathSuitSkill, Suit
from ..formula import get_base_attributes, 获取唤醒属性
from ..roleinfo import CharacterEmblemInfo, CharacterEquipInfo, get_key_by_value, CharacterOathInfo
@@ -178,6 +178,7 @@ class CharacterBase(CharacterProperty):
self.skill_names_cache = {}
self.skills_cache = {}
self.load_skills()
self.skill_effects = SkillEffectEngine(self)
# region 角色属性设置
@@ -188,6 +189,7 @@ class CharacterBase(CharacterProperty):
for index, i in enumerate([get_key_by_value(i) for i in ['力量', '体力', '智力', '精神']]):
setattr(self, i, getattr(self, i) + temp[index] + 唤醒)
@skill_effect_batch
def SetStatus(
self,
力量=0,
@@ -323,6 +325,7 @@ class CharacterBase(CharacterProperty):
value = int(value * (1 + self.jade_effect.ElementIncrease))
self.ElementDB[element] += value
@skill_effect_batch
def AddSkillLv(self, min: int, max: int, lv: int, type=-1, exceptSkills: tuple[str, ...] = ()) -> None:
"""
增加技能等级
@@ -334,18 +337,21 @@ class CharacterBase(CharacterProperty):
if (skillType == 'all' or skill.type == skillType) and skill.lv > 0 and skill.name not in exceptSkills:
skill.lv += lv
@skill_effect_batch
def SetSkillCD(self, min=1, max=100, cd=0, exclude: tuple[int, ...] = (50, 85, 100)) -> None:
"""设置技能CD"""
for skill in self.skills:
if min <= skill.learnLv <= max and (skill.type == 'active' or skill.damage) and skill.learnLv not in exclude:
skill.cdReduce *= 1 - cd
@skill_effect_batch
def SetSkillCDRecover(self, min=1, max=100, cd=0, exclude: tuple[int, ...] = (50, 85, 100)) -> None:
"""设置技能CD恢复"""
for skill in self.skills:
if min <= skill.learnLv <= max and (skill.type == 'active' or skill.damage) and skill.learnLv not in exclude:
skill.cdRecover += cd
@skill_effect_batch
def SetSkillRation(self, min=1, max=100, ratio=1, type=-1) -> None:
"""设置技能倍率
0 修改技能面板 1 不修改技能面板
+18 -13
View File
@@ -8,6 +8,7 @@ from ..roleinfo import CharacterEmblemInfo, CharacterEquipInfo, CharacterOathInf
from .base import CharacterBase
from .config_view import CalcConfigView
from .effect_apply import apply_uppercase_attrs
from ..skill_effects import skill_effect_batch
@cache
@@ -60,8 +61,9 @@ class CalcEquipMixin(CharacterBase):
self.SetDetail(setInfo)
self.SetSkills(setInfo.get('skills', {}))
@skill_effect_batch
def SetSkills(self, skills: dict[str, dict]):
"""设置技能等级"""
"""批量设置基础等级及 VP/UP;等级不含关联加级,共享等级由 level_source 决定。"""
for key in skills:
skill = self.GetSkillByID(key)
if skill is not None:
@@ -69,7 +71,7 @@ class CalcEquipMixin(CharacterBase):
skill.up = skills[key].get('up', 0)
if skill.hasVP:
skill.vp = skills[key].get('vp', 0)
skill.lv += skills[key].get('lv', 0)
self.skill_effects.set_input_level(skill, skills[key].get('lv', 0))
def load_skills(self) -> None:
"""加载技能"""
@@ -436,17 +438,20 @@ class CalcEquipMixin(CharacterBase):
def calc(self, setInfo: dict[str, dict]):
config = CalcConfigView(setInfo).as_legacy_dict()
self.calc_init(config)
self.SetBaseStatus()
self.calc_jade(config.get('jades', []))
self.calc_avatar(config.get('avatar', {}))
self.calc_basic()
self.calc_sundry(config.get('sundry', {}))
self.calc_emblems()
self.calc_oath_upgrade()
suit = self.calc_suits()
self.calc_equs()
self.calc_oaths()
# 装备等外部加成作为技能关联的基础值;批次结束后按最终技能状态重建关联。
# 装备之间仍有先后依赖(如先计算套装积分),不能据此任意交换下面的调用。
with self.skill_effects.batch():
self.calc_init(config)
self.SetBaseStatus()
self.calc_jade(config.get('jades', []))
self.calc_avatar(config.get('avatar', {}))
self.calc_basic()
self.calc_sundry(config.get('sundry', {}))
self.calc_emblems()
self.calc_oath_upgrade()
suit = self.calc_suits()
self.calc_equs()
self.calc_oaths()
skillInfos = []
if self.buffer:
skillInfos = self.calc_buffer_skills()
+71 -12
View File
@@ -8,6 +8,7 @@ from .formula import 武器冷却惩罚
from database.connect import get_redis_client
from api.core.Redis import get_redis_info
from api.utils.openapi_data import OPENAPI_CUSTOM_CN_PATH, logical_skill_detail_path
from .skill_effects import CHARACTER_ATTRIBUTES, INPUT_ATTRIBUTES
if TYPE_CHECKING:
from core.basic.character import Character
@@ -84,9 +85,21 @@ def get_data(prefix: str, key: int | str, func: Callable[[Any], Any] = _identity
class Skill:
def __setattr__(self, name, value):
char = self.__dict__.get('char')
engine = getattr(char, 'skill_effects', None)
if name != 'lv' and name in INPUT_ATTRIBUTES and engine is not None and engine.owns(self) and not engine.resolving:
engine.set_attribute(self, name, value)
else:
object.__setattr__(self, name, value)
name: str = ''
"""技能名称"""
_lv: int = 0
level_source: str | None = None
"""共享等级的源技能;目标不独立加级,也不使用自身 maxLv 截断。"""
association_source: bool | None = None
"""共享等级技能默认仅复用等级;需要独立提供关联效果时显式开启。"""
id: int = 0
"""技能ID"""
icon: str = '$char/$uuid'
@@ -131,15 +144,15 @@ class Skill:
cdRecover: float = 1
"""技能冷却恢复"""
associate: list = []
"""关联技能
type: 关联类型 *乘算 +加算 =直接赋值 $角色属性 默认'*skillDamage'
e.g.
*skillDamage 乘算技能伤害
*cdReduce 乘算技能冷却缩减
$*PATK 乘算角色物理攻击
~直接走type对应的函数名称 传参为old,new,data,skills,exceptSkills
skills: 关联技能 默认 '*'
exceptSkills: 例外技能 默认[]
"""关联声明;维护示例和扩展边界见 core/basic/skill_effects.py 文件顶部说明。
type 默认 '*skillDamage':*乘算、+加算、=覆盖;$*PAtkP 等作用于角色。
skills 默认 '*',表示所有伤害技能;显式列表使用技能名称,可指定被动技能。
exceptSkills 默认 [];weapon 默认职业武器选项;precondition() 控制前置条件。
data 按来源等级取值;ratio 是数据除数,默认 100,与技能的 ratio 属性不同。
+lv 的整数加级通常设置 ratio=1;=覆盖直接使用 data 值,不除以 ratio。
动态 effect 以 (0, 最终等级) 多次收集,只构造声明并调用 super().effect。
自定义 type 方法仍有兼容入口,但直接修改目标会绕过贡献收集,新增规则勿照搬。
"""
mode: list[str] = ['']
currentMode: str = ''
@@ -159,6 +172,10 @@ class Skill:
def __init__(self, char):
self.char = char
# Shared-level display skills override the setter with a no-op. They
# did not apply their own associations in the original calculation.
if self.association_source is None:
self.association_source = self.level_source is None and type(self).lv.fset is Skill.lv.fset
self.id = self._extract_id()
self.icon = self._generate_icon()
if self.line is None:
@@ -199,11 +216,22 @@ class Skill:
@property
def lv(self):
"""技能等级"""
engine = getattr(self.char, 'skill_effects', None)
if engine is not None and engine.owns(self):
engine.read_level(self)
elif self.level_source is not None:
return self.char.GetSkillByName(self.level_source).lv
return self._lv
@lv.setter
def lv(self, value):
value = int(min(self.maxLv, value))
if self.level_source is not None:
return
engine = getattr(self.char, 'skill_effects', None)
if engine is not None and engine.owns(self) and not engine.resolving:
engine.set_level(self, max(0, int(value)))
return
value = int(max(0, min(self.maxLv, value)))
if self._lv != value:
self.effect(self._lv, value)
self._lv = value
@@ -238,10 +266,20 @@ class Skill:
value = (1 - self.get_data_value(data, new) / ratio) / (1 - self.get_data_value(data, old) / ratio)
else:
value = (1 + self.get_data_value(data, new) / ratio) / (1 + self.get_data_value(data, old) / ratio)
self.char.SetStatus(**{type[2:]: value})
engine = getattr(self.char, 'skill_effects', None)
if engine is not None and engine.collecting is self:
if type[2:] in CHARACTER_ATTRIBUTES:
engine.record(self, self.char, type[2:], '*', value)
else:
self.char.SetStatus(**{type[2:]: value})
elif type.startswith('$+'):
value = self.get_data_value(data, new) / ratio - self.get_data_value(data, old) / ratio
self.char.SetStatus(**{type[2:]: value})
engine = getattr(self.char, 'skill_effects', None)
if engine is not None and engine.collecting is self:
if type[2:] in CHARACTER_ATTRIBUTES:
engine.record(self, self.char, type[2:], '+', value)
else:
self.char.SetStatus(**{type[2:]: value})
else:
exceptSkills = set(exceptSkills)
if skills == '*':
@@ -256,6 +294,10 @@ class Skill:
skill = self.char.GetSkillByName(name)
if skill is not None:
self.update_skill_attribute(skill, type, old, new, data, ratio, weapon, weapon_type, weapon_options)
else:
engine = getattr(self.char, 'skill_effects', None)
if engine is not None and engine.collecting is self:
engine.missing_target(self, name)
except Exception as e:
# print(f"Error applying association {type} for skill {self.name}: {e}")
raise e
@@ -265,6 +307,23 @@ class Skill:
return
old_value = self.get_data_value(data, old)
new_value = self.get_data_value(data, new)
engine = getattr(self.char, 'skill_effects', None)
if engine is not None and engine.collecting is self:
if type[1:] == 'lv' and skill.level_source is not None:
return
if type.startswith('*'):
if 'cdReduce' in type or 'cdRatio' in type:
value = (1 - new_value / ratio) / (1 - old_value / ratio)
else:
value = (1 + new_value / ratio) / (1 + old_value / ratio)
elif type.startswith('+'):
if type[1:] == 'lv' and not engine.is_learned(skill):
return
value = (new_value - old_value) / ratio
else:
value = new_value
engine.record(self, skill, type[1:], type[0], value)
return
if type.startswith('*'):
if 'cdReduce' in type or type == 'cd' or 'cdRatio' in type:
setattr(skill, type[1:], getattr(skill, type[1:]) * (1 - new_value / ratio) / (1 - old_value / ratio))
+386
View File
@@ -0,0 +1,386 @@
"""技能关联计算与维护说明
技能关联由 `core/basic/skill_effects.py` 的 `SkillEffectEngine` 统一处理。
`lv`、VP 等输入变化不再通过历史倍率乘除更新目标,而是重新解析最终状态并汇总本次关联贡献。
## 计算阶段
1. `SetSkills` 批量读取基础等级、VP、UP。传入的等级是基础等级,不含关联技能提供的等级;重复传入同一配置不会重复加级。
2. 解析等级和参数依赖。`lv`、`ratio`、`maxLayer` 是状态属性;关联到这些属性的规则以及 `level_source` 组成依赖图,按拓扑顺序求值。关联规则读取其他技能等级时,依赖会自动记录。循环、缺失共享等级源、计算过程中改变状态依赖会报出具体技能。
3. 基于最终状态收集伤害、冷却和角色属性贡献,并在装备提供的基础值之上应用一次。
`calc` 在处理配置、装备、时装、徽章等阶段使用同一个批次。需要读取等级时,可以先解析状态;伤害及角色属性关联在准备阶段结束后统一生效。
## 后续在哪里维护
| 新增或修改的影响 | 维护位置 | 要点 |
| --- | --- | --- |
| 技伤、冷却、攻击段数等普通关联 | 对应 `core/character/GF/职业/角色.py` 的技能类 `associate` | 通常不改引擎;目标属性应有默认值 |
| 随 VP、等级或自身参数变化的关联 | 来源技能的 `effect` | 每次生成完整声明,再调用 `super().effect(old, new)` |
| 技能给其他技能加级 | 来源技能的 `associate`,使用 `+lv` | 整数加级设置 `ratio=1`;只给已学习的目标加级 |
| 生命之树这类共享等级 | 目标技能的 `level_source` | 指向源技能名称,不再写等级 getter/no-op setter |
| 新增会被关联修改、并影响后续关联的参数 | 技能类默认值 + `core/basic/skill_effects.py` 的 `STATE_ATTRIBUTES` | 例如现有 `ratio`、`maxLayer`;加入后自动属于输入属性 |
| 新增外部配置输入,改变它需要重新计算关联 | 技能类、配置赋值入口 + `INPUT_ATTRIBUTES` | 例如现有 `vp`;单纯登记不提供跨技能参数读取依赖 |
| 新增角色属性关联 | `skill.py` 的角色贡献收集分支、`CHARACTER_ATTRIBUTES`,并核对 `character/base.py` 的 `SetStatus` | 当前仅支持对应标量字段;特殊运算需单独实现 |
| 新增运算符、目标选择方式、跨技能非等级依赖 | `skill.py` 的公共关联入口 + `skill_effects.py` | 属于公共能力扩展,需配套单元测试 |
| 装备改变技能等级或倍率 | 原装备规则及 `character/base.py` 的公共修改方法 | 使用已有批处理方法;低层直接写入须在 `batch()` 内 |
| 仅影响单次技能展示/输出的模式、VP、UP 变换 | 原技能 `setMode`、`vp_N`、输出计算方法 | `skillInfo()` 使用临时副本,不要误改为持久关联 |
普通职业规则优先放在职业文件。只有新增公共能力时才修改引擎;不要再用属性 setter 手动撤销旧倍率并乘入新倍率。
## 声明示例和数值约定
以下片段放在对应技能类内,`data0` 为该来源技能的等级数据表:
```python
# 默认百分比:data0[lv] = 110 时,倍率为 1 + 110 / 100。
associate = [{'type': '*skillDamage', 'data': data0, 'skills': ['目标技能名称']}]
# 加级:data0[lv] = 1 时,给已学习目标增加 1 级。
associate = [{'type': '+lv', 'data': data0, 'ratio': 1, 'skills': ['目标技能名称']}]
```
上面是两种独立示例;需要同时生效时放进同一列表。`skills='*'` 仅选择伤害技能,关联被动技能时使用明确的名称列表。名称必须与目标 `name` 完全一致,当前不是按 UUID 匹配。
数据以来源技能等级索引,通常令 `data[0] = 0`。收集仍兼容旧公式:普通乘算为 `(1 + data[lv]/ratio) / (1 + data[0]/ratio)`,`cdReduce`、`cdRatio` 改用减号;加算为 `(data[lv] - data[0])/ratio`;覆盖 `=` 直接取 `data[lv]`。这里字典中的 `ratio` 是除数,区别于手雷精通类上的 `ratio` 参数。
女弹药的动态关联是现成参考:
```python
ratio = 1.0
def effect(self, old, new):
self.associate = [{
'type': '*skillDamage',
'data': [value * self.ratio for value in self.data0],
'skills': ['G-14手雷', 'G-35L感电手雷', 'G-18C冰冻手雷'],
}]
return super().effect(old, new)
```
三觉被动通过 `*ratio` 关联手雷精通。引擎先确定最终 `ratio=1.2`,再收集手雷倍率,所以精通 11 级对应 `1 + 110% × 1.2 = 2.32`。
## 职业规则
静态 `associate` 格式不变。现有动态 `effect(old, new)` 作为兼容入口,在收集模式下以 `(0, 最终等级)` 执行;公共 `apply_association` / `update_skill_attribute` 记录贡献,不修改目标。每次收集都从原始 `associate` 声明开始,取消 VP 时不会保留上一次分支。
动态规则只能生成关联声明,不能在收集过程中直接修改角色或目标技能。需要修改关联产生方式时,应改用声明,而不是另写参数 setter 撤销旧倍率。元素师原先的 `ratio` 补偿 setter 已移除,避免与公共引擎重复应用。
引擎在创建时保存初始声明和状态默认值。动态分支应新建并赋值 `self.associate`,不要对共享的类级列表执行 `append` 或原地改写;也不要把运行后修改 `associate` 当作注册新规则。一次重算可能多次调用 `effect`,因此不能在其中累计计数、直接调用角色加成方法或依赖随机结果。
依赖图目前自动识别三类依赖:共享等级源、收集时跨技能 `.lv` 读取、指向状态属性的关联。
直接读取另一个技能的 `.ratio` 等非等级参数不会自动建立依赖。
新增这类规则时需要先扩展公共依赖机制并验证顺序;单独加入 `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/smoke/test_calc_availability.py -k configuration_order -q
```
第二项验证技能配置顺序(包括搭配武器的场景),不验证装备执行顺序。修改公共引擎后再运行完整 `backend/tests` 回归。
"""
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
# 会被关联修改、且参与后续关联求值的属性在此登记,并在技能类上提供默认值。
STATE_ATTRIBUTES = frozenset({'lv', 'ratio', 'maxLayer'})
# 外部赋值触发重算的输入;登记在此不等于自动跟踪跨技能的该属性读取。
INPUT_ATTRIBUTES = STATE_ATTRIBUTES | {'vp', 'up', 'comboCol', 'stage', 'currentMode'}
# These are the scalar keys actually supported by CharacterBase.SetStatus.
# 扩展前核对 SetStatus 的实际运算语义;特殊字段不能仅加入白名单。
CHARACTER_ATTRIBUTES = frozenset({
'STR', 'INT', 'Spirit', 'Vitality', 'AtkP', 'AtkM', 'AtkI',
'PAtkP', 'PAtkM', 'PAtkI', 'PSTR', 'PINT',
})
_MISSING = object()
@dataclass(frozen=True)
class SkillContribution:
source: object
index: int
target: object
attribute: str
operation: str
value: float
def skill_effect_batch(method):
"""Keep external skill/stat changes outside the association baseline."""
@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}
self._output_base = {}
self._depth = 0
self.resolving = False
self.collecting = None
self._pending = []
self._reads = set()
self._dirty = True
self._state_dirty = True
self.contributions = ()
self.unresolved_targets = set()
def owns(self, skill):
# skillInfo uses shallow copies; these are local render contexts.
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):
if skill.level_source is not None:
return
# += equipment bonuses must change the input, not copy already-derived
# mastery levels back into it. Input configuration uses 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):
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))
def _collect(self, skill):
# 一个重算周期会多次调用 effect;职业实现只能构造声明并调用公共收集入口。
self._pending = []
self._reads = set()
if not skill.association_source or skill.lv <= 0:
return (), set()
# VP branches often only assign associate when enabled. Start from the
# declaration every time so disabling VP also removes the old branch.
skill.associate = self._associations[skill]
self.collecting = skill
try:
skill.effect(0, skill.lv)
return tuple(self._pending), self._reads.copy()
finally:
self.collecting = None
@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 的关联。
# 其他跨技能属性读取不会自动建边;新增此类规则须先扩展依赖机制。
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 = self.character.GetSkillByName(skill.level_source)
if source is None:
raise ValueError(f'共享等级源技能不存在: {skill.name} -> {skill.level_source}')
graph[skill].add(source)
contributions, reads = self._collect(skill)
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:
# All upstream level/parameter contributions are ready now.
if skill.level_source is not None:
skill._lv = self.character.GetSkillByName(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()