yexuejc/cultivation-world-simulator
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor regions

Browse Source
This commit is contained in:
bridge
2025-09-10 22:55:31 +08:00
parent 6425e80ffe
commit 12fdccfee5
24 changed files with 664 additions and 438 deletions
Download Patch File Download Diff File Expand all files Collapse all files

351
src/run/create_map.py
View File

@@ -15,7 +15,7 @@ def create_cultivation_world_map() -> Map:
_create_base_terrain(game_map)
# 创建区域
_create_regions(game_map)
_assign_regions_to_tiles(game_map)
return game_map
@@ -208,347 +208,20 @@ def _add_other_terrains(game_map: Map):
if game_map.tiles[(x, y)].type == TileType.PLAIN:
game_map.tiles[(x, y)].type = TileType.SWAMP
def _create_regions(game_map: Map):
"""创建区域并分配灵气"""
# 收集各地形的坐标
terrain_coords = {}
def _assign_regions_to_tiles(game_map: Map):
"""将区域分配给地图中的tiles"""
# 初始化所有tiles的region为None
for x in range(game_map.width):
for y in range(game_map.height):
tile_type = game_map.tiles[(x, y)].type
if tile_type not in terrain_coords:
terrain_coords[tile_type] = []
terrain_coords[tile_type].append((x, y))
game_map.tiles[(x, y)].region = None
# 西域流沙 (大漠)
if TileType.DESERT in terrain_coords:
essence = Essence({
EssenceType.FIRE: 8,
EssenceType.EARTH: 7,
EssenceType.GOLD: 6,
EssenceType.WATER: 1,
EssenceType.WOOD: 2
})
game_map.create_region(
"西域流沙",
"茫茫大漠，黄沙漫天。此地火行灵气浓郁，土行次之，乃是修炼火系功法的绝佳之地。",
essence,
terrain_coords[TileType.DESERT]
)
# 南疆蛮荒 (雨林)
if TileType.RAINFOREST in terrain_coords:
essence = Essence({
EssenceType.WOOD: 8, # 木行主属性，但低于洞府的10
EssenceType.WATER: 6,
EssenceType.EARTH: 5,
EssenceType.FIRE: 3,
EssenceType.GOLD: 2
})
game_map.create_region(
"南疆蛮荒",
"古木参天，藤蔓缠绕。此地木行灵气极为浓郁，水行次之，是修炼木系功法和炼制灵药的宝地。",
essence,
terrain_coords[TileType.RAINFOREST]
)
# 极北冰原 (冰原)
if TileType.GLACIER in terrain_coords:
essence = Essence({
EssenceType.WATER: 8,
EssenceType.GOLD: 6,
EssenceType.EARTH: 4,
EssenceType.FIRE: 1,
EssenceType.WOOD: 3
})
game_map.create_region(
"极北冰原",
"千里冰封，万年不化。此地水行灵气充沛，金行次之，寒气逼人，唯有修炼寒冰功法者方可久居。",
essence,
terrain_coords[TileType.GLACIER]
)
# 无边碧海 (海洋)
if TileType.SEA in terrain_coords:
essence = Essence({
EssenceType.WATER: 8, # 水行主属性，但低于洞府的10
EssenceType.GOLD: 4,
EssenceType.WOOD: 3,
EssenceType.EARTH: 2,
EssenceType.FIRE: 1
})
game_map.create_region(
"无边碧海",
"浩瀚无垠，波涛汹涌。此地水行灵气达到极致，蕴含无穷玄机，是水系修士向往的圣地。",
essence,
terrain_coords[TileType.SEA]
)
# 天河奔流 (大河)
if TileType.WATER in terrain_coords:
essence = Essence({
EssenceType.WATER: 7,
EssenceType.WOOD: 5,
EssenceType.EARTH: 4,
EssenceType.GOLD: 3,
EssenceType.FIRE: 2
})
game_map.create_region(
"天河奔流",
"一江春水向东流，奔腾不息入东海。此河贯穿东西，水行灵气丰沛，滋养两岸万物生灵。",
essence,
terrain_coords[TileType.WATER]
)
# 青峰山脉 (山脉)
if TileType.MOUNTAIN in terrain_coords:
essence = Essence({
EssenceType.EARTH: 8,
EssenceType.GOLD: 7,
EssenceType.FIRE: 5,
EssenceType.WOOD: 3,
EssenceType.WATER: 2
})
game_map.create_region(
"青峰山脉",
"连绵起伏，直插云霄。此地土行灵气深厚，金行次之，乃是修炼土系功法和寻找天材地宝的胜地。",
essence,
terrain_coords[TileType.MOUNTAIN]
)
# 万丈雪峰 (雪山)
if TileType.SNOW_MOUNTAIN in terrain_coords:
essence = Essence({
EssenceType.WATER: 7, # 水行主属性，但低于洞府的10
EssenceType.GOLD: 6, # 金行次要，但低于洞府的10
EssenceType.EARTH: 6,
EssenceType.FIRE: 1,
EssenceType.WOOD: 2
})
game_map.create_region(
"万丈雪峰",
"雪峰皑皑，寒风刺骨。此地水行与金行灵气并重，是修炼冰系神通和淬炼法宝的绝佳之地。",
essence,
terrain_coords[TileType.SNOW_MOUNTAIN]
)
# 碧野千里 (草原)
if TileType.GRASSLAND in terrain_coords:
essence = Essence({
EssenceType.WOOD: 5, # 木行属性适中
EssenceType.EARTH: 5, # 土行属性适中
EssenceType.WATER: 5,
EssenceType.GOLD: 3,
EssenceType.FIRE: 4
})
game_map.create_region(
"碧野千里",
"芳草萋萋，一望无际。此地木土并重，灵气平和，是修炼基础功法和放牧灵兽的理想之地。",
essence,
terrain_coords[TileType.GRASSLAND]
)
# 青云林海 (森林)
if TileType.FOREST in terrain_coords:
essence = Essence({
EssenceType.WOOD: 7, # 木行主属性，但低于洞府的10
EssenceType.WATER: 4,
EssenceType.EARTH: 4,
EssenceType.GOLD: 3,
EssenceType.FIRE: 3
})
game_map.create_region(
"青云林海",
"古树参天，绿意盎然。此地木行灵气浓郁，是修炼木系功法、采集灵草和驯服林间灵兽的宝地。",
essence,
terrain_coords[TileType.FOREST]
)
# 炎狱火山 (火山)
if TileType.VOLCANO in terrain_coords:
essence = Essence({
EssenceType.FIRE: 8, # 火行主属性，但低于洞府的10
EssenceType.EARTH: 7,
EssenceType.GOLD: 4,
EssenceType.WATER: 1,
EssenceType.WOOD: 1
})
game_map.create_region(
"炎狱火山",
"烈焰冲天，岩浆奔流。此地火行灵气达到极致，是修炼火系神通和锻造法宝的圣地，常人不可近。",
essence,
terrain_coords[TileType.VOLCANO]
)
# 为每个2*2城市、洞穴和遗迹创建独立区域
_create_city_regions(game_map)
_create_wuxing_caves_regions(game_map)
_create_ruins_regions(game_map)
# 沃土良田 (农田)
if TileType.FARM in terrain_coords:
essence = Essence({
EssenceType.WOOD: 6, # 木行属性较强
EssenceType.EARTH: 6, # 土行属性较强
EssenceType.WATER: 6,
EssenceType.GOLD: 2,
EssenceType.FIRE: 3
})
game_map.create_region(
"沃土良田",
"土地肥沃，五谷丰登。此地木土并重，水行充沛，是种植灵药和培育灵植的理想之地。",
essence,
terrain_coords[TileType.FARM]
)
# 平原地带 (平原)
if TileType.PLAIN in terrain_coords:
essence = Essence({
EssenceType.EARTH: 5,
EssenceType.WOOD: 4,
EssenceType.WATER: 4,
EssenceType.GOLD: 3,
EssenceType.FIRE: 3
})
game_map.create_region(
"平原地带",
"地势平坦，灵气平和。此地五行均衡，是初学修炼者打基础和建立宗门的理想之地。",
essence,
terrain_coords[TileType.PLAIN]
)
# 迷雾沼泽 (沼泽)
if TileType.SWAMP in terrain_coords:
essence = Essence({
EssenceType.WATER: 6, # 水行属性较强
EssenceType.WOOD: 5, # 木行属性适中
EssenceType.EARTH: 5,
EssenceType.FIRE: 2,
EssenceType.GOLD: 3
})
game_map.create_region(
"迷雾沼泽",
"雾气缭绕，泥泞不堪。此地水木灵气浓郁，但瘴气丛生，是修炼毒功和寻找奇异灵草的危险之地。",
essence,
terrain_coords[TileType.SWAMP]
)
def _create_city_regions(game_map: Map):
"""为每个2*2城市创建独立区域"""
cities = [
{"name": "青云城", "base_x": 34, "base_y": 21, "description": "繁华都市，人烟稠密，商贾云集。此地金行灵气较为集中，是交易天材地宝、寻找机缘的重要场所。"},
{"name": "沙月城", "base_x": 14, "base_y": 19, "description": "沙漠绿洲中的贸易重镇，各路商队在此集结。金行灵气充沛，是修士补给和交流的重要据点。"},
{"name": "翠林城", "base_x": 54, "base_y": 14, "description": "森林深处的修仙重镇，木行灵气与金行灵气并重。众多修士在此栖居，是修炼和炼宝的理想之地。"}
]
for city in cities:
base_x, base_y = city["base_x"], city["base_y"]
city_coords = []
for dx in range(2):
for dy in range(2):
x, y = base_x + dx, base_y + dy
if game_map.is_in_bounds(x, y):
city_coords.append((x, y))
essence = Essence({
EssenceType.GOLD: 6,
EssenceType.EARTH: 5,
EssenceType.WOOD: 5,
EssenceType.WATER: 4,
EssenceType.FIRE: 4
})
game_map.create_region(
city["name"],
city["description"],
essence,
city_coords
)
def _create_wuxing_caves_regions(game_map: Map):
"""为每个2*2五行洞府创建独立区域"""
wuxing_caves = [
{"name": "太白金府", "base_x": 26, "base_y": 12, "element": EssenceType.GOLD,
"description": "青峰山脉深处的金行洞府，金精气凝，刀剑鸣音不绝，乃金系修士的最高圣地。"},
{"name": "青木洞天", "base_x": 48, "base_y": 18, "element": EssenceType.WOOD,
"description": "青云林海中的木行洞府，生机盎然，灵药遍地，乃木系修士的最高圣地。"},
{"name": "玄水秘境", "base_x": 67, "base_y": 25, "element": EssenceType.WATER,
"description": "无边碧海深处的水行洞府，碧波万里，水精凝神，乃水系修士的最高圣地。"},
{"name": "离火洞府", "base_x": 50, "base_y": 33, "element": EssenceType.FIRE,
"description": "炎狱火山旁的火行洞府，烈焰冲天，真火精纯，乃火系修士的最高圣地。"},
{"name": "厚土玄宫", "base_x": 30, "base_y": 16, "element": EssenceType.EARTH,
"description": "青峰山脉的土行洞府，厚德载物，山岳共鸣，乃土系修士的最高圣地。"}
]
for cave in wuxing_caves:
base_x, base_y = cave["base_x"], cave["base_y"]
cave_coords = []
for dx in range(2):
for dy in range(2):
x, y = base_x + dx, base_y + dy
if game_map.is_in_bounds(x, y):
cave_coords.append((x, y))
# 每个洞府的主属性灵气为10（最高值），其他属性较低
essence_config = {essence_type: 2 for essence_type in EssenceType}
essence_config[cave["element"]] = 10 # 主属性达到最高值
essence = Essence(essence_config)
game_map.create_region(
cave["name"],
cave["description"],
essence,
cave_coords
)
def _create_ruins_regions(game_map: Map):
"""为每个2*2遗迹创建独立区域"""
ruins = [
{"name": "古越遗迹", "base_x": 25, "base_y": 40, "description": "雨林深处的上古遗迹，古藤缠绕，木行灵气与金行灵气交融。蕴藏古老功法与灵药配方。"},
{"name": "沧海遗迹", "base_x": 66, "base_y": 47, "description": "沉没在海中的远古文明遗迹，水行灵气浓郁，潮汐间偶有宝物现世。"}
]
for ruin in ruins:
base_x, base_y = ruin["base_x"], ruin["base_y"]
ruin_coords = []
for dx in range(2):
for dy in range(2):
x, y = base_x + dx, base_y + dy
if game_map.is_in_bounds(x, y):
ruin_coords.append((x, y))
# 根据遗迹位置调整灵气配置
if ruin["name"] == "古越遗迹": # 雨林深处
essence = Essence({
EssenceType.WOOD: 8,
EssenceType.GOLD: 6,
EssenceType.WATER: 5,
EssenceType.EARTH: 4,
EssenceType.FIRE: 3
})
else: # 沧海遗迹，海中
essence = Essence({
EssenceType.WATER: 9,
EssenceType.GOLD: 6,
EssenceType.EARTH: 3,
EssenceType.WOOD: 3,
EssenceType.FIRE: 2
})
game_map.create_region(
ruin["name"],
ruin["description"],
essence,
ruin_coords
)
# 遍历所有region，为对应的坐标分配正确的region
# 现在从map实例获取region信息，而不是直接从region.py导入
for region in game_map.regions.values():
for coord_x, coord_y in region.cors:
# 确保坐标在地图范围内
if game_map.is_in_bounds(coord_x, coord_y):
game_map.tiles[(coord_x, coord_y)].region = region
if __name__ == "__main__":
# 创建地图

198
src/run/log.py Normal file
View File

@@ -0,0 +1,198 @@
"""
通用日志模块
功能：
1. 自动记录各种调用的输入和输出（目前主要用于LLM）
2. 启动时自动删除一周以前的日志文件
3. 日志文件按日期分组
"""
import logging
import os
import json
import time
from datetime import datetime, timedelta
from pathlib import Path
from typing import Optional
class Logger:
"""通用日志记录器"""
def __init__(self, log_dir: str = "logs"):
"""
初始化日志记录器
Args:
log_dir: 日志文件存储目录
"""
self.log_dir = Path(log_dir)
self.log_dir.mkdir(exist_ok=True)
# 清理旧日志文件
self._cleanup_old_logs()
# 设置当前日志文件
self._setup_current_logger()
def _cleanup_old_logs(self):
"""删除一周以前的日志文件"""
cutoff_date = datetime.now() - timedelta(days=7)
for log_file in self.log_dir.glob("*.log"):
try:
# 从文件名中提取日期
date_str = log_file.stem.split("_")[-1] # 取最后一部分作为日期
file_date = datetime.strptime(date_str, "%Y%m%d")
if file_date < cutoff_date:
log_file.unlink()
print(f"已删除过期日志文件: {log_file}")
except (ValueError, OSError) as e:
print(f"处理日志文件 {log_file} 时出错: {e}")
def _setup_current_logger(self):
"""设置当前日期的日志记录器"""
current_date = datetime.now().strftime("%Y%m%d")
log_filename = f"app_{current_date}.log"
self.log_file_path = self.log_dir / log_filename
# 创建日志记录器
self.logger = logging.getLogger(f"app_logger_{current_date}")
self.logger.setLevel(logging.INFO)
# 清除现有的处理器（避免重复记录）
self.logger.handlers.clear()
# 创建文件处理器
handler = logging.FileHandler(
self.log_file_path,
encoding='utf-8',
mode='a'
)
# 设置日志格式
formatter = logging.Formatter(
'%(asctime)s - %(levelname)s - %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
)
handler.setFormatter(formatter)
self.logger.addHandler(handler)
# 不向根日志记录器传播
self.logger.propagate = False
def log_llm_interaction(self,
model_name: str,
prompt: str,
response: str,
duration: Optional[float] = None,
additional_info: Optional[dict] = None):
"""
记录LLM交互
Args:
model_name: 使用的模型名称
prompt: 输入的提示词
response: LLM的响应
duration: 调用耗时（秒）
additional_info: 额外信息
"""
log_data = {
"timestamp": datetime.now().isoformat(),
"model_name": model_name,
"prompt": prompt,
"response": response,
"prompt_length": len(prompt),
"response_length": len(response),
"duration": duration
}
if additional_info:
log_data.update(additional_info)
# 记录日志
log_message = f"LLM_INTERACTION: {json.dumps(log_data, ensure_ascii=False)}"
self.logger.info(log_message)
def log_error(self, error_message: str, prompt: str = None):
"""
记录错误
Args:
error_message: 错误信息
prompt: 相关的提示词（可选）
"""
log_data = {
"timestamp": datetime.now().isoformat(),
"error": error_message,
"prompt": prompt if prompt else None
}
log_message = f"LLM_ERROR: {json.dumps(log_data, ensure_ascii=False)}"
self.logger.error(log_message)
def get_today_stats(self) -> dict:
"""
获取今日统计信息
Returns:
dict: 包含今日调用次数、总耗时等信息
"""
if not self.log_file_path.exists():
return {
"total_calls": 0,
"total_duration": 0,
"total_prompt_length": 0,
"total_response_length": 0,
"errors": 0
}
stats = {
"total_calls": 0,
"total_duration": 0,
"total_prompt_length": 0,
"total_response_length": 0,
"errors": 0
}
with open(self.log_file_path, 'r', encoding='utf-8') as f:
for line in f:
if "LLM_INTERACTION:" in line:
try:
json_str = line.split("LLM_INTERACTION: ", 1)[1]
data = json.loads(json_str)
stats["total_calls"] += 1
stats["total_duration"] += data.get("duration", 0) or 0
stats["total_prompt_length"] += data.get("prompt_length", 0)
stats["total_response_length"] += data.get("response_length", 0)
except (json.JSONDecodeError, IndexError):
pass
elif "LLM_ERROR:" in line:
stats["errors"] += 1
return stats
# 全局日志记录器实例
_logger = None
def get_logger() -> Logger:
"""获取全局日志记录器实例"""
global _logger
if _logger is None:
_logger = Logger()
return _logger
# LLM专用的便捷函数
def log_llm_call(model_name: str, prompt: str, response: str, duration: float = None):
"""便捷函数：记录LLM调用"""
logger = get_logger()
logger.log_llm_interaction(model_name, prompt, response, duration)
def log_llm_error(error_message: str, prompt: str = None):
"""便捷函数：记录LLM错误"""
logger = get_logger()
logger.log_error(error_message, prompt)
# 向后兼容的别名
get_llm_logger = get_logger

5
src/run/run.py
View File

@@ -23,6 +23,7 @@ from src.run.create_map import create_cultivation_world_map
from src.utils.names import get_random_name
from src.utils.id_generator import get_avatar_id
from src.utils.config import CONFIG
from src.run.log import get_logger
def clamp(value: int, lo: int, hi: int) -> int:
@@ -90,6 +91,10 @@ def make_avatars(world: World, count: int = 12, current_month_stamp: MonthStamp
async def main():
# 为了每次更丰富，使用随机种子；如需复现可将 seed 固定
# 初始化日志系统（会自动清理旧日志）
logger = get_logger()
print(f"日志系统已初始化，日志文件：{logger.log_file_path}")
game_map = create_cultivation_world_map()
world = World(map=game_map, month_stamp=create_month_stamp(Year(100), Month.JANUARY))