From 57cf5ca51a9b380acc65f7478eb1767f2e737ff3 Mon Sep 17 00:00:00 2001 From: bridge Date: Sun, 11 Jan 2026 23:53:26 +0800 Subject: [PATCH] add history class --- src/classes/history.py | 175 +++++++++++++++++++++++++ static/templates/history_influence.txt | 53 ++++++++ tests/test_history.py | 129 ++++++++++++++++++ 3 files changed, 357 insertions(+) create mode 100644 src/classes/history.py create mode 100644 static/templates/history_influence.txt create mode 100644 tests/test_history.py diff --git a/src/classes/history.py b/src/classes/history.py new file mode 100644 index 0000000..07c283d --- /dev/null +++ b/src/classes/history.py @@ -0,0 +1,175 @@ +import asyncio +import json +from pathlib import Path +from typing import Dict, Any, Optional, TYPE_CHECKING +import logging + +from src.classes.item_registry import ItemRegistry +from src.classes.technique import techniques_by_id, techniques_by_name +from src.classes.weapon import weapons_by_name +from src.utils.llm.client import call_llm_with_template +from src.run.log import get_logger + +if TYPE_CHECKING: + from src.classes.world import World + +class HistoryManager: + """ + 历史管理器 + 在游戏开局时，根据历史文本一次性修改世界中的对象数据。 + """ + def __init__(self, world: "World"): + self.world = world + # 配置目录路径 + self.config_dir = Path("static/game_configs") + self.logger = get_logger().logger + + async def apply_history_influence(self, history_text: str): + """ + 核心方法：读取 CSV -> LLM 分析 -> 更新内存对象 + """ + # 1. 准备 Prompt 参数：直接读取 CSV 原始内容 + infos = { + "world_info": str(self.world.static_info) if self.world else "", + "history_str": history_text, + "city_regions": self._read_csv("city_region.csv"), + "normal_regions": self._read_csv("normal_region.csv"), + "cultivate_regions": self._read_csv("cultivate_region.csv"), + "techniques": self._read_csv("technique.csv"), + "weapons": self._read_csv("weapon.csv"), + "auxiliarys": self._read_csv("auxiliary.csv"), + } + + # 2. 调用 LLM + self.logger.info("[History] 正在根据历史推演世界变化...") + try: + result = await call_llm_with_template( + template_path="static/templates/history_influence.txt", + infos=infos, + max_retries=3 # 增加重试次数，确保 JSON 格式正确 + ) + except Exception as e: + self.logger.error(f"[History] LLM 调用或解析失败: {e}") + return + + # 3. 应用变更到内存对象 + if result: + self._apply_changes(result) + else: + self.logger.info("[History] LLM 返回为空，未进行任何修改") + + def _read_csv(self, filename: str) -> str: + """读取 CSV 文件原始内容""" + file_path = self.config_dir / filename + if not file_path.exists(): + self.logger.warning(f"[History] Warning: 配置文件不存在 {file_path}") + return "" + try: + return file_path.read_text(encoding='utf-8') + except Exception as e: + self.logger.error(f"[History] 读取文件 {filename} 失败: {e}") + return "" + + def _apply_changes(self, result: Dict[str, Any]): + """分发并应用变更""" + + # 3.1 区域变更 + self._update_regions(result.get("city_regions_change", {})) + self._update_regions(result.get("normal_regions_change", {})) + self._update_regions(result.get("cultivate_regions_change", {})) + + # 3.2 功法变更 + self._update_techniques(result.get("techniques_change", {})) + + # 3.3 装备变更 + self._update_items(result.get("weapons_change", {}), weapons_by_name) + self._update_items(result.get("auxiliarys_change", {}), None) # 辅助装备可能没有全局 name 索引 + + def _update_regions(self, changes: Dict[str, Any]): + """更新区域 (Map.regions)""" + if not changes: return + + count = 0 + for rid_str, data in changes.items(): + try: + rid = int(rid_str) + # 从 World.Map 获取区域 + if self.world and self.world.map: + region = self.world.map.regions.get(rid) + if region: + self._update_obj_attrs(region, data) + self.logger.info(f"[History] 区域变更 - ID: {rid}, Name: {region.name}, Desc: {region.desc}") + count += 1 + except Exception as e: + self.logger.error(f"[History] 区域更新失败 - ID: {rid_str}, Error: {e}") + continue + if count > 0: + self.logger.info(f"[History] 更新了 {count} 个区域") + + def _update_techniques(self, changes: Dict[str, Any]): + """更新功法 (techniques_by_id)""" + if not changes: return + + count = 0 + for tid_str, data in changes.items(): + try: + tid = int(tid_str) + tech = techniques_by_id.get(tid) + if tech: + old_name = tech.name + self._update_obj_attrs(tech, data) + + # 同步 techniques_by_name 索引 + if tech.name != old_name: + if old_name in techniques_by_name: + del techniques_by_name[old_name] + techniques_by_name[tech.name] = tech + + self.logger.info(f"[History] 功法变更 - ID: {tid}, Name: {tech.name}, Desc: {tech.desc}") + count += 1 + except Exception as e: + self.logger.error(f"[History] 功法更新失败 - ID: {tid_str}, Error: {e}") + continue + if count > 0: + self.logger.info(f"[History] 更新了 {count} 本功法") + + def _update_items(self, changes: Dict[str, Any], by_name_index: Optional[Dict[str, Any]]): + """更新物品 (ItemRegistry)""" + if not changes: return + + count = 0 + for iid_str, data in changes.items(): + try: + iid = int(iid_str) + item = ItemRegistry.get(iid) + if item: + old_name = item.name + self._update_obj_attrs(item, data) + + # 同步可选的 name 索引 (如 weapons_by_name) + if by_name_index is not None and item.name != old_name: + if old_name in by_name_index: + del by_name_index[old_name] + by_name_index[item.name] = item + + self.logger.info(f"[History] 装备变更 - ID: {iid}, Name: {item.name}, Desc: {item.desc}") + count += 1 + except Exception as e: + self.logger.error(f"[History] 装备更新失败 - ID: {iid_str}, Error: {e}") + continue + if count > 0: + self.logger.info(f"[History] 更新了 {count} 件装备") + + def _update_obj_attrs(self, obj: Any, data: Dict[str, Any]): + """通用属性更新 helper""" + if "name" in data and data["name"]: + obj.name = str(data["name"]) + if "desc" in data and data["desc"]: + obj.desc = str(data["desc"]) + +if __name__ == "__main__": + # 模拟运行 + history_str = "上古时期..." + # 注意：这里直接运行可能会报错，因为需要 World 对象 + # 这里只是为了保留文件结构的完整性 + pass diff --git a/static/templates/history_influence.txt b/static/templates/history_influence.txt new file mode 100644 index 0000000..1e7ccf3 --- /dev/null +++ b/static/templates/history_influence.txt @@ -0,0 +1,53 @@ +你是一个仙侠世界的创作者，我会给你一个原始的世界背景，以及一段历史。 +你需要基于世界背景，根据这段历史，修改这个世界中存在的部分功法、兵器、辅助装备、区域信息。 + +世界背景： +{world_info} + +历史文本： +{history_str} + +城市区域信息： +{city_regions} + +普通区域信息： +{normal_regions} + +修炼区域信息： +{cultivate_regions} + +功法信息： +{techniques} + +兵器信息： +{weapons} + +辅助装备信息: +{auxiliarys} + +基于以上信息，分析，并返回修改意见。 + +返回JSON格式： +{{ + "thinking": "分析应该有怎么样的修改", + "city_regions_change": + {{ + "id": {{ //原来的id + "name": str // 新的名字 + "desc": desc // 新的desc + }} + }}, + "normal_regions_change": {{}} // dict, 结构同上 + "cultivate_regions_change": {{}} // dict, 结构同上 + "techniques_change": {{}} // dict, 结构同上 + "weapons_change": {{}} // dict, 结构同上 + "auxiliarys_change": {{}} // dict, 结构同上 +}} + +要求： +1. thinking是你的思考过程，要详细分析 +2. 要参考history的内容进行修改，言之有理，比如。 +3. history的文本内容多，就多修改点，少的话就少修改点 +4. 某项没有修改的话，就返回空字典{{}} +5. 要修改的项，只返回name和desc，不返回别的key + diff --git a/tests/test_history.py b/tests/test_history.py new file mode 100644 index 0000000..0b5ec46 --- /dev/null +++ b/tests/test_history.py @@ -0,0 +1,129 @@ +import pytest +from unittest.mock import MagicMock, AsyncMock, patch +from src.classes.history import HistoryManager +from src.classes.region import CityRegion, NormalRegion, CultivateRegion +from src.classes.technique import Technique, TechniqueAttribute, TechniqueGrade +from src.classes.weapon import Weapon, WeaponType +from src.classes.auxiliary import Auxiliary +from src.classes.cultivation import Realm +from src.classes.item_registry import ItemRegistry + +# 假设这些全局字典在模块层级 +from src.classes import technique as technique_module +from src.classes import weapon as weapon_module +# auxiliary 模块没有导出全局字典，所以这里不需要特别处理它的全局字典，只需要处理 ItemRegistry + +@pytest.mark.asyncio +async def test_history_influence(base_world): + # --- Setup Test Data --- + + # 1. Regions + city_region = CityRegion(id=1, name="OldCity", desc="Old Desc") + normal_region = NormalRegion(id=2, name="OldWild", desc="Old Wild Desc") + cult_region = CultivateRegion(id=3, name="OldCave", desc="Old Cave Desc") + + base_world.map.regions = { + 1: city_region, + 2: normal_region, + 3: cult_region + } + + # 2. Techniques + tech = Technique( + id=101, + name="OldTech", + attribute=TechniqueAttribute.GOLD, + grade=TechniqueGrade.LOWER, + desc="Old Tech Desc", + weight=1.0, + condition="" + ) + + # 3. Weapons & Auxiliaries + weapon = Weapon( + id=201, + name="OldSword", + weapon_type=WeaponType.SWORD, + realm=Realm.Qi_Refinement, + desc="Old Sword Desc" + ) + aux = Auxiliary( + id=301, + name="OldOrb", + realm=Realm.Qi_Refinement, + desc="Old Orb Desc" + ) + + # --- Patch Global Registries --- + # 使用 patch.dict 来隔离对全局字典的修改 + with patch.dict(technique_module.techniques_by_id, {101: tech}, clear=True), \ + patch.dict(technique_module.techniques_by_name, {"OldTech": tech}, clear=True), \ + patch.dict(weapon_module.weapons_by_name, {"OldSword": weapon}, clear=True), \ + patch.object(ItemRegistry, "_items_by_id", {201: weapon, 301: aux}): # ItemRegistry 是类属性 + + # --- Prepare LLM Mock Response --- + mock_response = { + "city_regions_change": { + "1": {"name": "NewCity", "desc": "New Desc"} + }, + "normal_regions_change": { + "2": {"name": "NewWild", "desc": "New Wild Desc"} + }, + "cultivate_regions_change": { + "3": {"name": "NewCave", "desc": "New Cave Desc"} + }, + "techniques_change": { + "101": {"name": "NewTech", "desc": "New Tech Desc"} + }, + "weapons_change": { + "201": {"name": "NewSword", "desc": "New Sword Desc"} + }, + "auxiliarys_change": { + "301": {"name": "NewOrb", "desc": "New Orb Desc"} + } + } + + # --- Instantiate Manager & Mock Internal Methods --- + manager = HistoryManager(base_world) + + # Mock _read_csv to return dummy string + manager._read_csv = MagicMock(return_value="dummy,csv,content") + + # Mock call_llm_with_template + with patch("src.classes.history.call_llm_with_template", new_callable=AsyncMock) as mock_llm: + mock_llm.return_value = mock_response + + # --- Execute --- + await manager.apply_history_influence("Some history text") + + # --- Assertions --- + + # 1. LLM Called + mock_llm.assert_called_once() + + # 2. Regions Updated + assert city_region.name == "NewCity" + assert city_region.desc == "New Desc" + assert normal_region.name == "NewWild" + assert normal_region.desc == "New Wild Desc" + assert cult_region.name == "NewCave" + assert cult_region.desc == "New Cave Desc" + + # 3. Technique Updated & Index Synced + assert tech.name == "NewTech" + assert tech.desc == "New Tech Desc" + assert "NewTech" in technique_module.techniques_by_name + assert "OldTech" not in technique_module.techniques_by_name + assert technique_module.techniques_by_name["NewTech"] == tech + + # 4. Weapon Updated & Index Synced + assert weapon.name == "NewSword" + assert weapon.desc == "New Sword Desc" + assert "NewSword" in weapon_module.weapons_by_name + assert "OldSword" not in weapon_module.weapons_by_name + assert weapon_module.weapons_by_name["NewSword"] == weapon + + # 5. Auxiliary Updated + assert aux.name == "NewOrb" + assert aux.desc == "New Orb Desc" +