feat: 重构配置加载器,增加配置文件版本控制和程序兼容能力
This commit is contained in:
@@ -6,12 +6,16 @@ import tomli
|
|||||||
import sys
|
import sys
|
||||||
from loguru import logger
|
from loguru import logger
|
||||||
from nonebot import get_driver
|
from nonebot import get_driver
|
||||||
|
from packaging import version
|
||||||
|
from packaging.version import Version, InvalidVersion
|
||||||
|
from packaging.specifiers import SpecifierSet,InvalidSpecifier
|
||||||
|
|
||||||
|
|
||||||
@dataclass
|
@dataclass
|
||||||
class BotConfig:
|
class BotConfig:
|
||||||
"""机器人配置类"""
|
"""机器人配置类"""
|
||||||
|
INNER_VERSION: SpecifierSet
|
||||||
|
|
||||||
BOT_QQ: Optional[int] = 1
|
BOT_QQ: Optional[int] = 1
|
||||||
BOT_NICKNAME: Optional[str] = None
|
BOT_NICKNAME: Optional[str] = None
|
||||||
|
|
||||||
@@ -78,130 +82,281 @@ class BotConfig:
|
|||||||
os.makedirs(config_dir)
|
os.makedirs(config_dir)
|
||||||
return config_dir
|
return config_dir
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def convert_to_specifierset(value: str) -> SpecifierSet:
|
||||||
|
"""将 字符串 版本表达式转换成 SpecifierSet
|
||||||
|
Args:
|
||||||
|
value[str]: 版本表达式(字符串)
|
||||||
|
Returns:
|
||||||
|
SpecifierSet
|
||||||
|
"""
|
||||||
|
|
||||||
|
try:
|
||||||
|
converted = SpecifierSet(value)
|
||||||
|
except InvalidSpecifier as e:
|
||||||
|
logger.error(
|
||||||
|
f"{value} 分类使用了错误的版本约束表达式\n",
|
||||||
|
"请阅读 https://semver.org/lang/zh-CN/ 修改代码"
|
||||||
|
)
|
||||||
|
exit(1)
|
||||||
|
|
||||||
|
return converted
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def get_config_version(cls, toml: dict) -> Version:
|
||||||
|
"""提取配置文件的 SpecifierSet 版本数据
|
||||||
|
Args:
|
||||||
|
toml[dict]: 输入的配置文件字典
|
||||||
|
Returns:
|
||||||
|
Version
|
||||||
|
"""
|
||||||
|
|
||||||
|
if 'inner' in toml:
|
||||||
|
try:
|
||||||
|
config_version : str = toml["inner"]["version"]
|
||||||
|
except KeyError as e:
|
||||||
|
logger.error(f"配置文件中 inner 段 不存在 {e}, 这是错误的配置文件")
|
||||||
|
exit(1)
|
||||||
|
else:
|
||||||
|
toml["inner"] = { "version": "0.0.0" }
|
||||||
|
config_version = toml["inner"]["version"]
|
||||||
|
|
||||||
|
try:
|
||||||
|
ver = version.parse(config_version)
|
||||||
|
except InvalidVersion as e:
|
||||||
|
logger.error(
|
||||||
|
"配置文件中 inner段 的 version 键是错误的版本描述\n"
|
||||||
|
"请阅读 https://semver.org/lang/zh-CN/ 修改配置,并参考本项目指定的模板进行修改\n"
|
||||||
|
"本项目在不同的版本下有不同的模板,请注意识别"
|
||||||
|
)
|
||||||
|
exit(1)
|
||||||
|
|
||||||
|
return ver
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def load_config(cls, config_path: str = None) -> "BotConfig":
|
def load_config(cls, config_path: str = None) -> "BotConfig":
|
||||||
"""从TOML配置文件加载配置"""
|
"""从TOML配置文件加载配置"""
|
||||||
config = cls()
|
config = cls()
|
||||||
if os.path.exists(config_path):
|
|
||||||
with open(config_path, "rb") as f:
|
|
||||||
toml_dict = tomli.load(f)
|
|
||||||
|
|
||||||
if 'personality' in toml_dict:
|
def personality(parent: dict):
|
||||||
personality_config=toml_dict['personality']
|
personality_config = parent['personality']
|
||||||
personality=personality_config.get('prompt_personality')
|
personality = personality_config.get('prompt_personality')
|
||||||
if len(personality) >= 2:
|
if len(personality) >= 2:
|
||||||
logger.info(f"载入自定义人格:{personality}")
|
logger.info(f"载入自定义人格:{personality}")
|
||||||
config.PROMPT_PERSONALITY=personality_config.get('prompt_personality',config.PROMPT_PERSONALITY)
|
config.PROMPT_PERSONALITY=personality_config.get('prompt_personality',config.PROMPT_PERSONALITY)
|
||||||
logger.info(f"载入自定义日程prompt:{personality_config.get('prompt_schedule',config.PROMPT_SCHEDULE_GEN)}")
|
logger.info(f"载入自定义日程prompt:{personality_config.get('prompt_schedule',config.PROMPT_SCHEDULE_GEN)}")
|
||||||
config.PROMPT_SCHEDULE_GEN=personality_config.get('prompt_schedule',config.PROMPT_SCHEDULE_GEN)
|
config.PROMPT_SCHEDULE_GEN=personality_config.get('prompt_schedule',config.PROMPT_SCHEDULE_GEN)
|
||||||
|
|
||||||
if "emoji" in toml_dict:
|
def emoji(parent: dict):
|
||||||
emoji_config = toml_dict["emoji"]
|
emoji_config = parent["emoji"]
|
||||||
config.EMOJI_CHECK_INTERVAL = emoji_config.get("check_interval", config.EMOJI_CHECK_INTERVAL)
|
config.EMOJI_CHECK_INTERVAL = emoji_config.get("check_interval", config.EMOJI_CHECK_INTERVAL)
|
||||||
config.EMOJI_REGISTER_INTERVAL = emoji_config.get("register_interval", config.EMOJI_REGISTER_INTERVAL)
|
config.EMOJI_REGISTER_INTERVAL = emoji_config.get("register_interval", config.EMOJI_REGISTER_INTERVAL)
|
||||||
config.EMOJI_CHECK_PROMPT = emoji_config.get('check_prompt',config.EMOJI_CHECK_PROMPT)
|
config.EMOJI_CHECK_PROMPT = emoji_config.get('check_prompt',config.EMOJI_CHECK_PROMPT)
|
||||||
config.EMOJI_SAVE = emoji_config.get('auto_save',config.EMOJI_SAVE)
|
config.EMOJI_SAVE = emoji_config.get('auto_save',config.EMOJI_SAVE)
|
||||||
config.EMOJI_CHECK = emoji_config.get('enable_check',config.EMOJI_CHECK)
|
config.EMOJI_CHECK = emoji_config.get('enable_check',config.EMOJI_CHECK)
|
||||||
|
|
||||||
if "cq_code" in toml_dict:
|
def cq_code(parent: dict):
|
||||||
cq_code_config = toml_dict["cq_code"]
|
cq_code_config = parent["cq_code"]
|
||||||
config.ENABLE_PIC_TRANSLATE = cq_code_config.get("enable_pic_translate", config.ENABLE_PIC_TRANSLATE)
|
config.ENABLE_PIC_TRANSLATE = cq_code_config.get("enable_pic_translate", config.ENABLE_PIC_TRANSLATE)
|
||||||
|
|
||||||
|
def bot(parent: dict):
|
||||||
# 机器人基础配置
|
# 机器人基础配置
|
||||||
if "bot" in toml_dict:
|
bot_config = parent["bot"]
|
||||||
bot_config = toml_dict["bot"]
|
|
||||||
bot_qq = bot_config.get("qq")
|
bot_qq = bot_config.get("qq")
|
||||||
config.BOT_QQ = int(bot_qq)
|
config.BOT_QQ = int(bot_qq)
|
||||||
config.BOT_NICKNAME = bot_config.get("nickname", config.BOT_NICKNAME)
|
config.BOT_NICKNAME = bot_config.get("nickname", config.BOT_NICKNAME)
|
||||||
|
|
||||||
if "response" in toml_dict:
|
def response(parent: dict):
|
||||||
response_config = toml_dict["response"]
|
response_config = parent["response"]
|
||||||
config.MODEL_R1_PROBABILITY = response_config.get("model_r1_probability", config.MODEL_R1_PROBABILITY)
|
config.MODEL_R1_PROBABILITY = response_config.get("model_r1_probability", config.MODEL_R1_PROBABILITY)
|
||||||
config.MODEL_V3_PROBABILITY = response_config.get("model_v3_probability", config.MODEL_V3_PROBABILITY)
|
config.MODEL_V3_PROBABILITY = response_config.get("model_v3_probability", config.MODEL_V3_PROBABILITY)
|
||||||
config.MODEL_R1_DISTILL_PROBABILITY = response_config.get("model_r1_distill_probability", config.MODEL_R1_DISTILL_PROBABILITY)
|
config.MODEL_R1_DISTILL_PROBABILITY = response_config.get("model_r1_distill_probability", config.MODEL_R1_DISTILL_PROBABILITY)
|
||||||
config.max_response_length = response_config.get("max_response_length", config.max_response_length)
|
config.max_response_length = response_config.get("max_response_length", config.max_response_length)
|
||||||
|
|
||||||
|
def model(parent: dict):
|
||||||
# 加载模型配置
|
# 加载模型配置
|
||||||
if "model" in toml_dict:
|
model_config = parent["model"]
|
||||||
model_config = toml_dict["model"]
|
config_version : Version = cls.get_config_version(parent)
|
||||||
|
|
||||||
if "llm_reasoning" in model_config:
|
config_list = [
|
||||||
config.llm_reasoning = model_config["llm_reasoning"]
|
"llm_reasoning",
|
||||||
|
"llm_reasoning_minor",
|
||||||
|
"llm_normal",
|
||||||
|
"llm_normal_minor",
|
||||||
|
"llm_topic_judge",
|
||||||
|
"llm_summary_by_topic",
|
||||||
|
"llm_emotion_judge",
|
||||||
|
"vlm",
|
||||||
|
"embedding",
|
||||||
|
"moderation"
|
||||||
|
]
|
||||||
|
|
||||||
if "llm_reasoning_minor" in model_config:
|
for item in config_list:
|
||||||
config.llm_reasoning_minor = model_config["llm_reasoning_minor"]
|
if item in model_config:
|
||||||
|
cfg_item = model_config[item]
|
||||||
|
|
||||||
if "llm_normal" in model_config:
|
# base_url 的例子: SILICONFLOW_BASE_URL
|
||||||
config.llm_normal = model_config["llm_normal"]
|
# key 的例子: SILICONFLOW_KEY
|
||||||
|
cfg_target = {
|
||||||
|
"name" : "",
|
||||||
|
"base_url" : "",
|
||||||
|
"key" : "",
|
||||||
|
"pri_in" : 0,
|
||||||
|
"pri_out" : 0
|
||||||
|
}
|
||||||
|
|
||||||
if "llm_normal_minor" in model_config:
|
if config_version in SpecifierSet("<0.0.0"):
|
||||||
config.llm_normal_minor = model_config["llm_normal_minor"]
|
cfg_target = cfg_item
|
||||||
|
|
||||||
if "llm_topic_judge" in model_config:
|
elif config_version in SpecifierSet(">=0.0.1"):
|
||||||
config.llm_topic_judge = model_config["llm_topic_judge"]
|
stable_item = ["name","pri_in","pri_out"]
|
||||||
|
for i in stable_item:
|
||||||
|
cfg_target[i] = cfg_item[i]
|
||||||
|
|
||||||
if "llm_summary_by_topic" in model_config:
|
provider = cfg_item["provider"]
|
||||||
config.llm_summary_by_topic = model_config["llm_summary_by_topic"]
|
|
||||||
|
|
||||||
if "llm_emotion_judge" in model_config:
|
cfg_target["base_url"] = f"{provider}_BASE_URL"
|
||||||
config.llm_emotion_judge = model_config["llm_emotion_judge"]
|
cfg_target["key"] = f"{provider}_KEY"
|
||||||
|
|
||||||
if "vlm" in model_config:
|
|
||||||
config.vlm = model_config["vlm"]
|
|
||||||
|
|
||||||
if "embedding" in model_config:
|
# 如果 列表中的项目在 model_config 中,利用反射来设置对应项目
|
||||||
config.embedding = model_config["embedding"]
|
setattr(config,item,cfg_target)
|
||||||
|
|
||||||
if "moderation" in model_config:
|
def message(parent: dict):
|
||||||
config.moderation = model_config["moderation"]
|
msg_config = parent["message"]
|
||||||
|
|
||||||
# 消息配置
|
|
||||||
if "message" in toml_dict:
|
|
||||||
msg_config = toml_dict["message"]
|
|
||||||
config.MIN_TEXT_LENGTH = msg_config.get("min_text_length", config.MIN_TEXT_LENGTH)
|
config.MIN_TEXT_LENGTH = msg_config.get("min_text_length", config.MIN_TEXT_LENGTH)
|
||||||
config.MAX_CONTEXT_SIZE = msg_config.get("max_context_size", config.MAX_CONTEXT_SIZE)
|
config.MAX_CONTEXT_SIZE = msg_config.get("max_context_size", config.MAX_CONTEXT_SIZE)
|
||||||
config.emoji_chance = msg_config.get("emoji_chance", config.emoji_chance)
|
config.emoji_chance = msg_config.get("emoji_chance", config.emoji_chance)
|
||||||
config.ban_words=msg_config.get("ban_words",config.ban_words)
|
config.ban_words=msg_config.get("ban_words",config.ban_words)
|
||||||
|
|
||||||
if "memory" in toml_dict:
|
def memory(parent: dict):
|
||||||
memory_config = toml_dict["memory"]
|
memory_config = parent["memory"]
|
||||||
config.build_memory_interval = memory_config.get("build_memory_interval", config.build_memory_interval)
|
config.build_memory_interval = memory_config.get("build_memory_interval", config.build_memory_interval)
|
||||||
config.forget_memory_interval = memory_config.get("forget_memory_interval", config.forget_memory_interval)
|
config.forget_memory_interval = memory_config.get("forget_memory_interval", config.forget_memory_interval)
|
||||||
|
|
||||||
if "mood" in toml_dict:
|
def mood(parent: dict):
|
||||||
mood_config = toml_dict["mood"]
|
mood_config = parent["mood"]
|
||||||
config.mood_update_interval = mood_config.get("mood_update_interval", config.mood_update_interval)
|
config.mood_update_interval = mood_config.get("mood_update_interval", config.mood_update_interval)
|
||||||
config.mood_decay_rate = mood_config.get("mood_decay_rate", config.mood_decay_rate)
|
config.mood_decay_rate = mood_config.get("mood_decay_rate", config.mood_decay_rate)
|
||||||
config.mood_intensity_factor = mood_config.get("mood_intensity_factor", config.mood_intensity_factor)
|
config.mood_intensity_factor = mood_config.get("mood_intensity_factor", config.mood_intensity_factor)
|
||||||
|
|
||||||
# 群组配置
|
def groups(parent: dict):
|
||||||
if "groups" in toml_dict:
|
groups_config = parent["groups"]
|
||||||
groups_config = toml_dict["groups"]
|
|
||||||
config.talk_allowed_groups = set(groups_config.get("talk_allowed", []))
|
config.talk_allowed_groups = set(groups_config.get("talk_allowed", []))
|
||||||
config.talk_frequency_down_groups = set(groups_config.get("talk_frequency_down", []))
|
config.talk_frequency_down_groups = set(groups_config.get("talk_frequency_down", []))
|
||||||
config.ban_user_id = set(groups_config.get("ban_user_id", []))
|
config.ban_user_id = set(groups_config.get("ban_user_id", []))
|
||||||
|
|
||||||
if "others" in toml_dict:
|
def others(parent: dict):
|
||||||
others_config = toml_dict["others"]
|
others_config = parent["others"]
|
||||||
config.enable_advance_output = others_config.get("enable_advance_output", config.enable_advance_output)
|
config.enable_advance_output = others_config.get("enable_advance_output", config.enable_advance_output)
|
||||||
config.enable_kuuki_read = others_config.get("enable_kuuki_read", config.enable_kuuki_read)
|
config.enable_kuuki_read = others_config.get("enable_kuuki_read", config.enable_kuuki_read)
|
||||||
|
|
||||||
|
# 版本表达式:>=1.0.0,<2.0.0
|
||||||
|
include_configs = {
|
||||||
|
"personality": {
|
||||||
|
"func": personality,
|
||||||
|
"support": ">=0.0.0"
|
||||||
|
},
|
||||||
|
"emoji": {
|
||||||
|
"func": emoji,
|
||||||
|
"support": ">=0.0.0"
|
||||||
|
},
|
||||||
|
"cq_code": {
|
||||||
|
"func": cq_code,
|
||||||
|
"support": ">=0.0.0"
|
||||||
|
},
|
||||||
|
"bot": {
|
||||||
|
"func": bot,
|
||||||
|
"support": ">=0.0.0"
|
||||||
|
},
|
||||||
|
"response": {
|
||||||
|
"func": response,
|
||||||
|
"support": ">=0.0.0"
|
||||||
|
},
|
||||||
|
"model": {
|
||||||
|
"func": model,
|
||||||
|
"support": ">=0.0.0"
|
||||||
|
},
|
||||||
|
"message": {
|
||||||
|
"func": message,
|
||||||
|
"support": ">=0.0.0"
|
||||||
|
},
|
||||||
|
"memory": {
|
||||||
|
"func": memory,
|
||||||
|
"support": ">=0.0.0"
|
||||||
|
},
|
||||||
|
"mood": {
|
||||||
|
"func": mood,
|
||||||
|
"support": ">=0.0.0"
|
||||||
|
},
|
||||||
|
"groups": {
|
||||||
|
"func": groups,
|
||||||
|
"support": ">=0.0.0"
|
||||||
|
},
|
||||||
|
"others": {
|
||||||
|
"func": others,
|
||||||
|
"support": ">=0.0.0"
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
# 原地修改,将 字符串版本表达式 转换成 版本对象
|
||||||
|
for key in include_configs:
|
||||||
|
item_support = include_configs[key]["support"]
|
||||||
|
include_configs[key]["support"] = cls.convert_to_specifierset(item_support)
|
||||||
|
|
||||||
|
if os.path.exists(config_path):
|
||||||
|
with open(config_path, "rb") as f:
|
||||||
|
toml_dict = tomli.load(f)
|
||||||
|
|
||||||
|
# 获取配置文件版本
|
||||||
|
config_version : Version = cls.get_config_version(toml_dict)
|
||||||
|
|
||||||
|
# 如果在配置中找到了需要的项,调用对应项的闭包函数处理
|
||||||
|
for key in include_configs:
|
||||||
|
if key in toml_dict:
|
||||||
|
group_specifierset: SpecifierSet = toml_dict[key]["support"]
|
||||||
|
|
||||||
|
# 检查配置文件版本是否在支持范围内
|
||||||
|
if config_version in group_specifierset:
|
||||||
|
# 如果版本在支持范围内,检查是否在支持的末端
|
||||||
|
if config_version == group_specifierset.filter([config_version])[-1]:
|
||||||
|
logger.warning(
|
||||||
|
f"配置文件中的 '{key}' 字段的版本 ({config_version}) 已接近支持范围的末端。\n"
|
||||||
|
f"未来版本可能会移除对该字段的支持。"
|
||||||
|
)
|
||||||
|
include_configs[key]["func"](toml_dict)
|
||||||
|
|
||||||
|
else:
|
||||||
|
# 如果版本不在支持范围内,崩溃并提示用户
|
||||||
|
logger.error(
|
||||||
|
f"配置文件中的 '{key}' 字段的版本 ({config_version}) 不在支持范围内。\n"
|
||||||
|
f"当前程序仅支持以下版本范围: {group_specifierset}"
|
||||||
|
)
|
||||||
|
exit(1)
|
||||||
|
|
||||||
|
else:
|
||||||
|
# 如果用户根本没有需要的配置项,提示缺少配置
|
||||||
|
logger.error(f"配置文件中缺少必需的字段: '{key}'")
|
||||||
|
exit(1)
|
||||||
|
|
||||||
logger.success(f"成功加载配置文件: {config_path}")
|
logger.success(f"成功加载配置文件: {config_path}")
|
||||||
|
|
||||||
return config
|
return config
|
||||||
|
|
||||||
# 获取配置文件路径
|
# 获取配置文件路径
|
||||||
|
|
||||||
bot_config_floder_path = BotConfig.get_config_dir()
|
bot_config_floder_path = BotConfig.get_config_dir()
|
||||||
print(f"正在品鉴配置文件目录: {bot_config_floder_path}")
|
print(f"正在品鉴配置文件目录: {bot_config_floder_path}")
|
||||||
|
|
||||||
bot_config_path = os.path.join(bot_config_floder_path, "bot_config.toml")
|
bot_config_path = os.path.join(bot_config_floder_path, "bot_config.toml")
|
||||||
|
|
||||||
if os.path.exists(bot_config_path):
|
if os.path.exists(bot_config_path):
|
||||||
# 如果开发环境配置文件不存在,则使用默认配置文件
|
# 如果开发环境配置文件不存在,则使用默认配置文件
|
||||||
print(f"异常的新鲜,异常的美味: {bot_config_path}")
|
print(f"异常的新鲜,异常的美味: {bot_config_path}")
|
||||||
logger.info("使用bot配置文件")
|
logger.info("使用bot配置文件")
|
||||||
else:
|
else:
|
||||||
logger.info("没有找到美味")
|
# 配置文件不存在
|
||||||
|
logger.error("配置文件不存在,请检查路径: {bot_config_path}")
|
||||||
|
raise FileNotFoundError(f"配置文件不存在: {bot_config_path}")
|
||||||
|
|
||||||
global_config = BotConfig.load_config(config_path=bot_config_path)
|
global_config = BotConfig.load_config(config_path=bot_config_path)
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user