feat: data reload system

This commit is contained in:
bridge
2026-01-18 16:37:08 +08:00
parent bd563b10f4
commit eb2c715069
7 changed files with 133 additions and 35 deletions

View File

@@ -68,16 +68,16 @@ class Auxiliary(Item):
}
def _load_auxiliaries() -> tuple[Dict[int, Auxiliary], Dict[str, Auxiliary]]:
def _load_auxiliaries_data() -> tuple[Dict[int, Auxiliary], Dict[str, Auxiliary]]:
"""从配表加载 auxiliary 数据。
返回:(按ID、按名称 的映射)。
返回:新的 (按ID、按名称 的映射)。
"""
auxiliaries_by_id: Dict[int, Auxiliary] = {}
auxiliaries_by_name: Dict[str, Auxiliary] = {}
new_by_id: Dict[int, Auxiliary] = {}
new_by_name: Dict[str, Auxiliary] = {}
df = game_configs.get("auxiliary")
if df is None:
return auxiliaries_by_id, auxiliaries_by_name
return new_by_id, new_by_name
for row in df:
effects = load_effect_from_str(get_str(row, "effects"))
@@ -100,17 +100,31 @@ def _load_auxiliaries() -> tuple[Dict[int, Auxiliary], Dict[str, Auxiliary]]:
effect_desc=effect_desc,
)
auxiliaries_by_id[a.id] = a
auxiliaries_by_name[a.name] = a
new_by_id[a.id] = a
new_by_name[a.name] = a
# 注册到全局注册表
from src.classes.item_registry import ItemRegistry
ItemRegistry.register(a.id, a)
return auxiliaries_by_id, auxiliaries_by_name
return new_by_id, new_by_name
# 全局容器
auxiliaries_by_id: Dict[int, Auxiliary] = {}
auxiliaries_by_name: Dict[str, Auxiliary] = {}
auxiliaries_by_id, auxiliaries_by_name = _load_auxiliaries()
def reload():
"""重新加载数据,保留全局字典引用"""
new_id, new_name = _load_auxiliaries_data()
auxiliaries_by_id.clear()
auxiliaries_by_id.update(new_id)
auxiliaries_by_name.clear()
auxiliaries_by_name.update(new_name)
# 模块初始化时执行一次
reload()
def get_random_auxiliary_by_realm(realm: Realm) -> Optional[Auxiliary]:

View File

@@ -22,3 +22,8 @@ class ItemRegistry:
@classmethod
def get_all(cls) -> Dict[int, "Item"]:
return cls._items_by_id
@classmethod
def reset(cls):
"""重置注册表,清空所有注册的物品"""
cls._items_by_id.clear()

View File

@@ -166,10 +166,12 @@ def _split_names(value: object) -> list[str]:
return parts
def _load_sects() -> tuple[dict[int, Sect], dict[str, Sect]]:
"""从配表加载 sect 数据"""
sects_by_id: dict[int, Sect] = {}
sects_by_name: dict[str, Sect] = {}
def _load_sects_data() -> tuple[dict[int, Sect], dict[str, Sect]]:
"""从配表加载 sect 数据
返回:新的 (sects_by_id, sects_by_name)
"""
new_by_id: dict[int, Sect] = {}
new_by_name: dict[str, Sect] = {}
df = game_configs["sect"]
# 读取宗门驻地映射(优先从 sect_region.csv 获取驻地地名/描述)
@@ -256,14 +258,27 @@ def _load_sects() -> tuple[dict[int, Sect], dict[str, Sect]]:
effect_desc=effect_desc,
rank_names=rank_names_map,
)
sects_by_id[sect.id] = sect
sects_by_name[sect.name] = sect
new_by_id[sect.id] = sect
new_by_name[sect.name] = sect
return sects_by_id, sects_by_name
return new_by_id, new_by_name
# 全局容器(保持引用不变)
sects_by_id: dict[int, Sect] = {}
sects_by_name: dict[str, Sect] = {}
# 导出:从配表加载 sect 数据
sects_by_id, sects_by_name = _load_sects()
def reload():
"""重新加载数据,保留全局字典引用"""
new_id, new_name = _load_sects_data()
sects_by_id.clear()
sects_by_id.update(new_id)
sects_by_name.clear()
sects_by_name.update(new_name)
# 模块初始化时执行一次
reload()
def get_sect_info_with_rank(avatar: "Avatar", detailed: bool = False) -> str:

View File

@@ -114,9 +114,12 @@ SUPPRESSION: dict[TechniqueAttribute, set[TechniqueAttribute]] = {
}
def loads() -> tuple[dict[int, Technique], dict[str, Technique]]:
techniques_by_id: dict[int, Technique] = {}
techniques_by_name: dict[str, Technique] = {}
def _load_techniques_data() -> tuple[dict[int, Technique], dict[str, Technique]]:
"""从配表加载 technique 数据
返回:新的 (techniques_by_id, techniques_by_name)
"""
new_by_id: dict[int, Technique] = {}
new_by_name: dict[str, Technique] = {}
df = game_configs["technique"]
for row in df:
attr = TechniqueAttribute(get_str(row, "technique_root"))
@@ -144,12 +147,27 @@ def loads() -> tuple[dict[int, Technique], dict[str, Technique]]:
effects=effects,
effect_desc=effect_desc,
)
techniques_by_id[t.id] = t
techniques_by_name[t.name] = t
return techniques_by_id, techniques_by_name
new_by_id[t.id] = t
new_by_name[t.name] = t
return new_by_id, new_by_name
techniques_by_id, techniques_by_name = loads()
# 全局容器(保持引用不变)
techniques_by_id: dict[int, Technique] = {}
techniques_by_name: dict[str, Technique] = {}
def reload():
"""重新加载数据,保留全局字典引用"""
new_id, new_name = _load_techniques_data()
techniques_by_id.clear()
techniques_by_id.update(new_id)
techniques_by_name.clear()
techniques_by_name.update(new_name)
# 模块初始化时执行一次
reload()
def is_attribute_compatible_with_root(attr: TechniqueAttribute, root: Root) -> bool:

View File

@@ -61,16 +61,16 @@ class Weapon(Item):
}
def _load_weapons() -> tuple[Dict[int, Weapon], Dict[str, Weapon]]:
def _load_weapons_data() -> tuple[Dict[int, Weapon], Dict[str, Weapon]]:
"""从配表加载 weapon 数据。
返回:(按ID、按名称 的映射)。
返回:新的 (按ID、按名称 的映射)。
"""
weapons_by_id: Dict[int, Weapon] = {}
weapons_by_name: Dict[str, Weapon] = {}
new_by_id: Dict[int, Weapon] = {}
new_by_name: Dict[str, Weapon] = {}
df = game_configs.get("weapon")
if df is None:
return weapons_by_id, weapons_by_name
return new_by_id, new_by_name
for row in df:
effects = load_effect_from_str(get_str(row, "effects"))
@@ -105,17 +105,31 @@ def _load_weapons() -> tuple[Dict[int, Weapon], Dict[str, Weapon]]:
effect_desc=effect_desc,
)
weapons_by_id[w.id] = w
weapons_by_name[w.name] = w
new_by_id[w.id] = w
new_by_name[w.name] = w
# 注册到全局注册表
# 注册到全局注册表 (注意ItemRegistry 在外部 reset 之后,这里会重新注册)
from src.classes.item_registry import ItemRegistry
ItemRegistry.register(w.id, w)
return weapons_by_id, weapons_by_name
return new_by_id, new_by_name
# 全局容器
weapons_by_id: Dict[int, Weapon] = {}
weapons_by_name: Dict[str, Weapon] = {}
weapons_by_id, weapons_by_name = _load_weapons()
def reload():
"""重新加载数据,保留全局字典引用"""
new_id, new_name = _load_weapons_data()
weapons_by_id.clear()
weapons_by_id.update(new_id)
weapons_by_name.clear()
weapons_by_name.update(new_name)
# 模块初始化时执行一次
reload()
def get_random_weapon_by_realm(realm: Realm, weapon_type: Optional[WeaponType] = None) -> Optional[Weapon]:

26
src/run/data_loader.py Normal file
View File

@@ -0,0 +1,26 @@
from src.classes.sect import reload as reload_sects
from src.classes.technique import reload as reload_techniques
from src.classes.weapon import reload as reload_weapons
from src.classes.auxiliary import reload as reload_auxiliaries
from src.classes.item_registry import ItemRegistry
from src.run.log import get_logger
def reload_all_static_data():
"""
重置所有游戏静态数据到初始状态。
必须在每次 init_game 之前调用。
"""
logger = get_logger().logger
logger.info("[DataLoader] 开始重置静态游戏数据...")
# 1. 清空物品注册表
ItemRegistry.reset()
# 2. 重新加载各模块数据
# 注意顺序有些模块可能依赖其他模块如功法可能依赖宗门ID虽通常只有弱引用
reload_sects()
reload_techniques()
reload_weapons()
reload_auxiliaries()
logger.info("[DataLoader] 静态数据重置完成,环境已净化。")

View File

@@ -43,6 +43,7 @@ import random
from omegaconf import OmegaConf
from src.utils.llm.client import test_connectivity
from src.utils.llm.config import LLMConfig, LLMMode
from src.run.data_loader import reload_all_static_data
# 全局游戏实例
game_instance = {
@@ -315,6 +316,11 @@ async def init_game_async():
try:
# 阶段 0: 资源扫描
update_init_progress(0, "scanning_assets")
# === 重置所有静态数据,清除历史修改污染 ===
print("正在重置世界规则数据...")
reload_all_static_data()
await asyncio.to_thread(scan_avatar_assets)
# 阶段 1: 地图加载