584 lines
23 KiB
Python
584 lines
23 KiB
Python
import os
|
||
import shutil
|
||
import sys
|
||
from datetime import datetime
|
||
|
||
import tomlkit
|
||
from pydantic import Field
|
||
from rich.traceback import install
|
||
from tomlkit import TOMLDocument
|
||
from tomlkit.items import KeyType, Table
|
||
|
||
from src.common.logger import get_logger
|
||
from src.config.config_base import ValidatedConfigBase
|
||
from src.config.official_configs import (
|
||
AffinityFlowConfig,
|
||
BotConfig,
|
||
ChatConfig,
|
||
ChineseTypoConfig,
|
||
CommandConfig,
|
||
CrossContextConfig,
|
||
CustomPromptConfig,
|
||
DatabaseConfig,
|
||
DebugConfig,
|
||
DependencyManagementConfig,
|
||
EmojiConfig,
|
||
ExperimentalConfig,
|
||
ExpressionConfig,
|
||
KokoroFlowChatterConfig,
|
||
LPMMKnowledgeConfig,
|
||
MessageBusConfig,
|
||
MemoryConfig,
|
||
MessageReceiveConfig,
|
||
MoodConfig,
|
||
NoticeConfig,
|
||
PermissionConfig,
|
||
PersonalityConfig,
|
||
PlanningSystemConfig,
|
||
PluginHttpSystemConfig,
|
||
ProactiveThinkingConfig,
|
||
ReactionConfig,
|
||
ResponsePostProcessConfig,
|
||
ResponseSplitterConfig,
|
||
ToolConfig,
|
||
VideoAnalysisConfig,
|
||
VoiceConfig,
|
||
WebSearchConfig,
|
||
)
|
||
|
||
from .api_ada_configs import (
|
||
APIProvider,
|
||
ModelInfo,
|
||
ModelTaskConfig,
|
||
)
|
||
|
||
install(extra_lines=3)
|
||
|
||
|
||
# 配置主程序日志格式
|
||
logger = get_logger("config")
|
||
|
||
# 获取当前文件所在目录的父目录的父目录（即MoFox-Bot项目根目录）
|
||
PROJECT_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))
|
||
CONFIG_DIR = os.path.join(PROJECT_ROOT, "config")
|
||
TEMPLATE_DIR = os.path.join(PROJECT_ROOT, "template")
|
||
|
||
# 考虑到，实际上配置文件中的mai_version是不会自动更新的,所以采用硬编码
|
||
# 对该字段的更新，请严格参照语义化版本规范：https://semver.org/lang/zh-CN/
|
||
MMC_VERSION = "0.13.1-alpha.1"
|
||
|
||
# 全局配置变量
|
||
_CONFIG_INITIALIZED = False
|
||
global_config: "Config | None" = None
|
||
model_config: "APIAdapterConfig | None" = None
|
||
|
||
|
||
def get_key_comment(toml_table, key):
|
||
# 获取key的注释（如果有）
|
||
if hasattr(toml_table, "trivia") and hasattr(toml_table.trivia, "comment"):
|
||
return toml_table.trivia.comment
|
||
if hasattr(toml_table, "value") and isinstance(toml_table.value, dict):
|
||
item = toml_table.value.get(key)
|
||
if item is not None and hasattr(item, "trivia"):
|
||
return item.trivia.comment
|
||
if hasattr(toml_table, "keys"):
|
||
for k in toml_table.keys():
|
||
if isinstance(k, KeyType) and k.key == key: # type: ignore
|
||
return k.trivia.comment # type: ignore
|
||
return None
|
||
|
||
|
||
def compare_dicts(new, old, path=None, logs=None):
|
||
# 递归比较两个dict，找出新增和删减项，收集注释
|
||
if path is None:
|
||
path = []
|
||
if logs is None:
|
||
logs = []
|
||
# 新增项
|
||
for key in new:
|
||
if key == "version":
|
||
continue
|
||
if key not in old:
|
||
comment = get_key_comment(new, key)
|
||
logs.append(f"新增: {'.'.join([*path, str(key)])} 注释: {comment or '无'}")
|
||
elif isinstance(new[key], dict | Table) and isinstance(old.get(key), dict | Table):
|
||
compare_dicts(new[key], old[key], [*path, str(key)], logs)
|
||
# 删减项
|
||
for key in old:
|
||
if key == "version":
|
||
continue
|
||
if key not in new:
|
||
comment = get_key_comment(old, key)
|
||
logs.append(f"删减: {'.'.join([*path, str(key)])} 注释: {comment or '无'}")
|
||
return logs
|
||
|
||
|
||
def get_value_by_path(d, path):
|
||
for k in path:
|
||
if isinstance(d, dict) and k in d:
|
||
d = d[k]
|
||
else:
|
||
return None
|
||
return d
|
||
|
||
|
||
def set_value_by_path(d, path, value):
|
||
for k in path[:-1]:
|
||
if k not in d or not isinstance(d[k], dict):
|
||
d[k] = {}
|
||
d = d[k]
|
||
d[path[-1]] = value
|
||
|
||
|
||
def compare_default_values(new, old, path=None, logs=None, changes=None):
|
||
# 递归比较两个dict，找出默认值变化项
|
||
if path is None:
|
||
path = []
|
||
if logs is None:
|
||
logs = []
|
||
if changes is None:
|
||
changes = []
|
||
for key in new:
|
||
if key == "version":
|
||
continue
|
||
if key in old:
|
||
if isinstance(new[key], dict | Table) and isinstance(old[key], dict | Table):
|
||
compare_default_values(new[key], old[key], [*path, str(key)], logs, changes)
|
||
elif new[key] != old[key]:
|
||
logs.append(f"默认值变化: {'.'.join([*path, str(key)])} 旧默认值: {old[key]} 新默认值: {new[key]}")
|
||
changes.append(([*path, str(key)], old[key], new[key]))
|
||
return logs, changes
|
||
|
||
|
||
def _get_version_from_toml(toml_path) -> str | None:
|
||
"""从TOML文件中获取版本号"""
|
||
if not os.path.exists(toml_path):
|
||
return None
|
||
with open(toml_path, encoding="utf-8") as f:
|
||
doc = tomlkit.load(f)
|
||
if "inner" in doc and "version" in doc["inner"]: # type: ignore
|
||
return doc["inner"]["version"] # type: ignore
|
||
return None
|
||
|
||
|
||
def _version_tuple(v):
|
||
"""将版本字符串转换为元组以便比较"""
|
||
if v is None:
|
||
return (0,)
|
||
return tuple(int(x) if x.isdigit() else 0 for x in str(v).replace("v", "").split("-")[0].split("."))
|
||
|
||
|
||
def _remove_obsolete_keys(target: TOMLDocument | dict | Table, reference: TOMLDocument | dict | Table):
|
||
"""
|
||
递归地从目标字典中移除所有不存在于参考字典中的键。
|
||
"""
|
||
# 使用 list() 创建键的副本，以便在迭代期间安全地修改字典
|
||
for key in list(target.keys()):
|
||
if key not in reference:
|
||
del target[key]
|
||
elif isinstance(target.get(key), dict | Table) and isinstance(reference.get(key), dict | Table):
|
||
_remove_obsolete_keys(target[key], reference[key]) # type: ignore
|
||
|
||
|
||
def _update_dict(target: TOMLDocument | dict | Table, source: TOMLDocument | dict):
|
||
"""
|
||
将source字典的值更新到target字典中
|
||
对于已存在的键，使用source的值进行更新
|
||
对于不存在的键，从source添加到target
|
||
"""
|
||
for key, value in source.items():
|
||
# 跳过version字段的更新
|
||
if key == "version":
|
||
continue
|
||
|
||
# 在合并 permission.master_users 时添加特别调试日志
|
||
if key == "permission" and isinstance(value, (dict, Table)) and "master_users" in value:
|
||
logger.info(f"【调试日志】在 _update_dict 中检测到 'permission' 表，其 'master_users' 的值为: {value['master_users']}")
|
||
|
||
|
||
if key in target:
|
||
# 键已存在，更新值
|
||
target_value = target[key]
|
||
if isinstance(value, dict) and isinstance(target_value, dict | Table):
|
||
_update_dict(target_value, value)
|
||
else:
|
||
try:
|
||
# 对数组类型进行特殊处理
|
||
if isinstance(value, list):
|
||
# 如果是空数组，确保它保持为空数组
|
||
target[key] = tomlkit.array(str(value)) if value else tomlkit.array()
|
||
else:
|
||
# 其他类型使用item方法创建新值
|
||
target[key] = tomlkit.item(value)
|
||
except (TypeError, ValueError):
|
||
# 如果转换失败，直接赋值
|
||
target[key] = value
|
||
else:
|
||
# 键不存在，从source添加新键到target
|
||
try:
|
||
if isinstance(value, dict):
|
||
# 对于字典类型，创建新的Table
|
||
new_table = tomlkit.table()
|
||
_update_dict(new_table, value)
|
||
target[key] = new_table
|
||
elif isinstance(value, list):
|
||
# 对于数组类型
|
||
target[key] = tomlkit.array(str(value)) if value else tomlkit.array()
|
||
else:
|
||
# 其他类型使用item方法创建新值
|
||
target[key] = tomlkit.item(value)
|
||
except (TypeError, ValueError):
|
||
# 如果转换失败，直接赋值
|
||
target[key] = value
|
||
|
||
|
||
def _update_config_generic(config_name: str, template_name: str):
|
||
"""
|
||
通用的配置文件更新函数
|
||
|
||
Args:
|
||
config_name: 配置文件名（不含扩展名），如 'bot_config' 或 'model_config'
|
||
template_name: 模板文件名（不含扩展名），如 'bot_config_template' 或 'model_config_template'
|
||
"""
|
||
# 获取根目录路径
|
||
old_config_dir = os.path.join(CONFIG_DIR, "old")
|
||
compare_dir = os.path.join(TEMPLATE_DIR, "compare")
|
||
|
||
# 定义文件路径
|
||
template_path = os.path.join(TEMPLATE_DIR, f"{template_name}.toml")
|
||
old_config_path = os.path.join(CONFIG_DIR, f"{config_name}.toml")
|
||
new_config_path = os.path.join(CONFIG_DIR, f"{config_name}.toml")
|
||
compare_path = os.path.join(compare_dir, f"{template_name}.toml")
|
||
|
||
# 创建compare目录（如果不存在）
|
||
os.makedirs(compare_dir, exist_ok=True)
|
||
|
||
template_version = _get_version_from_toml(template_path)
|
||
compare_version = _get_version_from_toml(compare_path)
|
||
|
||
# 检查配置文件是否存在
|
||
if not os.path.exists(old_config_path):
|
||
logger.info(f"{config_name}.toml配置文件不存在，从模板创建新配置")
|
||
os.makedirs(CONFIG_DIR, exist_ok=True) # 创建文件夹
|
||
shutil.copy2(template_path, old_config_path) # 复制模板文件
|
||
logger.info(f"已创建新{config_name}配置文件，请填写后重新运行: {old_config_path}")
|
||
# 新创建配置文件，退出
|
||
sys.exit(0)
|
||
|
||
compare_config = None
|
||
new_config = None
|
||
old_config = None
|
||
|
||
# 先读取 compare 下的模板（如果有），用于默认值变动检测
|
||
if os.path.exists(compare_path):
|
||
with open(compare_path, encoding="utf-8") as f:
|
||
compare_config = tomlkit.load(f)
|
||
|
||
# 读取当前模板
|
||
with open(template_path, encoding="utf-8") as f:
|
||
new_config = tomlkit.load(f)
|
||
|
||
# 检查默认值变化并处理（只有 compare_config 存在时才做）
|
||
if compare_config:
|
||
# 读取旧配置
|
||
with open(old_config_path, encoding="utf-8") as f:
|
||
old_config = tomlkit.load(f)
|
||
logs, changes = compare_default_values(new_config, compare_config)
|
||
if logs:
|
||
logger.info(f"检测到{config_name}模板默认值变动如下：")
|
||
for log in logs:
|
||
logger.info(log)
|
||
# 检查旧配置是否等于旧默认值，如果是则更新为新默认值
|
||
for path, old_default, new_default in changes:
|
||
old_value = get_value_by_path(old_config, path)
|
||
if old_value == old_default:
|
||
set_value_by_path(old_config, path, new_default)
|
||
logger.info(
|
||
f"已自动将{config_name}配置 {'.'.join(path)} 的值从旧默认值 {old_default} 更新为新默认值 {new_default}"
|
||
)
|
||
else:
|
||
logger.info(f"未检测到{config_name}模板默认值变动")
|
||
|
||
# 检查 compare 下没有模板，或新模板版本更高，则复制
|
||
if not os.path.exists(compare_path):
|
||
shutil.copy2(template_path, compare_path)
|
||
logger.info(f"已将{config_name}模板文件复制到: {compare_path}")
|
||
elif _version_tuple(template_version) > _version_tuple(compare_version):
|
||
shutil.copy2(template_path, compare_path)
|
||
logger.info(f"{config_name}模板版本较新，已替换compare下的模板: {compare_path}")
|
||
else:
|
||
logger.debug(f"compare下的{config_name}模板版本不低于当前模板，无需替换: {compare_path}")
|
||
|
||
# 读取旧配置文件和模板文件（如果前面没读过 old_config，这里再读一次）
|
||
if old_config is None:
|
||
with open(old_config_path, encoding="utf-8") as f:
|
||
old_config = tomlkit.load(f)
|
||
# new_config 已经读取
|
||
|
||
# 检查version是否相同
|
||
if old_config and "inner" in old_config and "inner" in new_config:
|
||
old_version = old_config["inner"].get("version") # type: ignore
|
||
new_version = new_config["inner"].get("version") # type: ignore
|
||
if old_version and new_version and old_version == new_version:
|
||
logger.info(f"检测到{config_name}配置文件版本号相同 (v{old_version})，跳过更新")
|
||
return
|
||
else:
|
||
logger.info(
|
||
f"\n----------------------------------------\n检测到{config_name}版本号不同: 旧版本 v{old_version} -> 新版本 v{new_version}\n----------------------------------------"
|
||
)
|
||
else:
|
||
logger.info(f"已有{config_name}配置文件未检测到版本号，可能是旧版本。将进行更新")
|
||
|
||
# 创建old目录（如果不存在）
|
||
os.makedirs(old_config_dir, exist_ok=True) # 生成带时间戳的新文件名
|
||
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||
old_backup_path = os.path.join(old_config_dir, f"{config_name}_{timestamp}.toml")
|
||
|
||
# 移动旧配置文件到old目录
|
||
shutil.move(old_config_path, old_backup_path)
|
||
logger.info(f"已备份旧{config_name}配置文件到: {old_backup_path}")
|
||
|
||
# 复制模板文件到配置目录
|
||
shutil.copy2(template_path, new_config_path)
|
||
logger.info(f"已创建新{config_name}配置文件: {new_config_path}")
|
||
|
||
# 输出新增和删减项及注释
|
||
if old_config:
|
||
logger.info(f"{config_name}配置项变动如下：\n----------------------------------------")
|
||
if logs := compare_dicts(new_config, old_config):
|
||
for log in logs:
|
||
logger.info(log)
|
||
else:
|
||
logger.info("无新增或删减项")
|
||
|
||
# 将旧配置的值更新到新配置中
|
||
logger.info(f"开始合并{config_name}新旧配置...")
|
||
_update_dict(new_config, old_config)
|
||
|
||
# 移除在新模板中已不存在的旧配置项
|
||
logger.info(f"开始移除{config_name}中已废弃的配置项...")
|
||
with open(template_path, encoding="utf-8") as f:
|
||
template_doc = tomlkit.load(f)
|
||
_remove_obsolete_keys(new_config, template_doc)
|
||
logger.info(f"已移除{config_name}中已废弃的配置项")
|
||
|
||
# 保存更新后的配置（保留注释和格式）
|
||
with open(new_config_path, "w", encoding="utf-8") as f:
|
||
f.write(tomlkit.dumps(new_config))
|
||
logger.info(f"{config_name}配置文件更新完成，建议检查新配置文件中的内容，以免丢失重要信息")
|
||
|
||
|
||
def update_config():
|
||
"""更新bot_config.toml配置文件"""
|
||
_update_config_generic("bot_config", "bot_config_template")
|
||
|
||
|
||
def update_model_config():
|
||
"""更新model_config.toml配置文件"""
|
||
_update_config_generic("model_config", "model_config_template")
|
||
|
||
|
||
class Config(ValidatedConfigBase):
|
||
"""总配置类"""
|
||
|
||
MMC_VERSION: str = Field(default=MMC_VERSION, description="MaiCore版本号")
|
||
|
||
database: DatabaseConfig = Field(..., description="数据库配置")
|
||
bot: BotConfig = Field(..., description="机器人基本配置")
|
||
personality: PersonalityConfig = Field(..., description="个性配置")
|
||
chat: ChatConfig = Field(..., description="聊天配置")
|
||
message_receive: MessageReceiveConfig = Field(..., description="消息接收配置")
|
||
notice: NoticeConfig = Field(..., description="Notice消息配置")
|
||
emoji: EmojiConfig = Field(..., description="表情配置")
|
||
expression: ExpressionConfig = Field(..., description="表达配置")
|
||
memory: MemoryConfig | None = Field(default=None, description="记忆配置")
|
||
mood: MoodConfig = Field(..., description="情绪配置")
|
||
reaction: ReactionConfig = Field(default_factory=ReactionConfig, description="反应规则配置")
|
||
chinese_typo: ChineseTypoConfig = Field(..., description="中文错别字配置")
|
||
response_post_process: ResponsePostProcessConfig = Field(..., description="响应后处理配置")
|
||
response_splitter: ResponseSplitterConfig = Field(..., description="响应分割配置")
|
||
experimental: ExperimentalConfig = Field(default_factory=lambda: ExperimentalConfig(), description="实验性功能配置")
|
||
message_bus: MessageBusConfig = Field(..., description="消息总线配置")
|
||
lpmm_knowledge: LPMMKnowledgeConfig = Field(..., description="LPMM知识配置")
|
||
tool: ToolConfig = Field(..., description="工具配置")
|
||
debug: DebugConfig = Field(..., description="调试配置")
|
||
custom_prompt: CustomPromptConfig = Field(..., description="自定义提示配置")
|
||
|
||
voice: VoiceConfig = Field(..., description="语音配置")
|
||
permission: PermissionConfig = Field(..., description="权限配置")
|
||
command: CommandConfig = Field(..., description="命令系统配置")
|
||
|
||
# 有默认值的字段放在后面
|
||
video_analysis: VideoAnalysisConfig = Field(
|
||
default_factory=lambda: VideoAnalysisConfig(), description="视频分析配置"
|
||
)
|
||
dependency_management: DependencyManagementConfig = Field(
|
||
default_factory=lambda: DependencyManagementConfig(), description="依赖管理配置"
|
||
)
|
||
web_search: WebSearchConfig = Field(default_factory=lambda: WebSearchConfig(), description="网络搜索配置")
|
||
planning_system: PlanningSystemConfig = Field(
|
||
default_factory=lambda: PlanningSystemConfig(), description="规划系统配置"
|
||
)
|
||
cross_context: CrossContextConfig = Field(
|
||
default_factory=lambda: CrossContextConfig(), description="跨群聊上下文共享配置"
|
||
)
|
||
affinity_flow: AffinityFlowConfig = Field(default_factory=lambda: AffinityFlowConfig(), description="亲和流配置")
|
||
proactive_thinking: ProactiveThinkingConfig = Field(
|
||
default_factory=lambda: ProactiveThinkingConfig(), description="主动思考配置"
|
||
)
|
||
kokoro_flow_chatter: KokoroFlowChatterConfig = Field(
|
||
default_factory=lambda: KokoroFlowChatterConfig(), description="心流对话系统配置（私聊专用）"
|
||
)
|
||
plugin_http_system: PluginHttpSystemConfig = Field(
|
||
default_factory=lambda: PluginHttpSystemConfig(), description="插件HTTP端点系统配置"
|
||
)
|
||
|
||
|
||
class APIAdapterConfig(ValidatedConfigBase):
|
||
"""API Adapter配置类"""
|
||
|
||
models: list[ModelInfo] = Field(..., min_length=1, description="模型列表")
|
||
model_task_config: ModelTaskConfig = Field(..., description="模型任务配置")
|
||
api_providers: list[APIProvider] = Field(..., min_length=1, description="API提供商列表")
|
||
|
||
def __init__(self, **data):
|
||
super().__init__(**data)
|
||
self.api_providers_dict = {provider.name: provider for provider in self.api_providers}
|
||
self.models_dict = {model.name: model for model in self.models}
|
||
|
||
@classmethod
|
||
def validate_models_list(cls, v):
|
||
"""验证模型列表"""
|
||
if not v:
|
||
raise ValueError("模型列表不能为空，请在配置中设置有效的模型列表。")
|
||
|
||
# 检查模型名称是否重复
|
||
model_names = [model.name for model in v]
|
||
if len(model_names) != len(set(model_names)):
|
||
raise ValueError("模型名称存在重复，请检查配置文件。")
|
||
|
||
# 检查模型标识符是否有效
|
||
for model in v:
|
||
if not model.model_identifier:
|
||
raise ValueError(f"模型 '{model.name}' 的 model_identifier 不能为空")
|
||
|
||
return v
|
||
|
||
@classmethod
|
||
def validate_api_providers_list(cls, v):
|
||
"""验证API提供商列表"""
|
||
if not v:
|
||
raise ValueError("API提供商列表不能为空，请在配置中设置有效的API提供商列表。")
|
||
|
||
# 检查API提供商名称是否重复
|
||
provider_names = [provider.name for provider in v]
|
||
if len(provider_names) != len(set(provider_names)):
|
||
raise ValueError("API提供商名称存在重复，请检查配置文件。")
|
||
|
||
return v
|
||
|
||
def get_model_info(self, model_name: str) -> ModelInfo:
|
||
"""根据模型名称获取模型信息"""
|
||
if not model_name:
|
||
raise ValueError("模型名称不能为空")
|
||
if model_name not in self.models_dict:
|
||
raise KeyError(f"模型 '{model_name}' 不存在")
|
||
return self.models_dict[model_name]
|
||
|
||
def get_provider(self, provider_name: str) -> APIProvider:
|
||
"""根据提供商名称获取API提供商信息"""
|
||
if not provider_name:
|
||
raise ValueError("API提供商名称不能为空")
|
||
if provider_name not in self.api_providers_dict:
|
||
raise KeyError(f"API提供商 '{provider_name}' 不存在")
|
||
return self.api_providers_dict[provider_name]
|
||
|
||
|
||
def load_config(config_path: str) -> Config:
|
||
"""
|
||
加载配置文件
|
||
Args:
|
||
config_path: 配置文件路径
|
||
Returns:
|
||
Config对象
|
||
"""
|
||
# 读取配置文件
|
||
with open(config_path, encoding="utf-8") as f:
|
||
config_data = tomlkit.load(f)
|
||
|
||
# 创建Config对象（各个配置类会自动进行 Pydantic 验证）
|
||
try:
|
||
logger.debug("正在解析和验证配置文件...")
|
||
config = Config.from_dict(config_data)
|
||
logger.debug("配置文件解析和验证完成")
|
||
|
||
# 【临时修复】在验证后，手动从原始数据重新加载 master_users
|
||
try:
|
||
# 先将 tomlkit 对象转换为纯 Python 字典
|
||
config_dict = config_data.unwrap()
|
||
if "permission" in config_dict and "master_users" in config_dict["permission"]:
|
||
raw_master_users = config_dict["permission"]["master_users"]
|
||
# 现在 raw_master_users 就是一个标准的 Python 列表了
|
||
config.permission.master_users = raw_master_users
|
||
logger.debug(f"【临时修复】已手动将 master_users 设置为: {config.permission.master_users}")
|
||
except Exception as patch_exc:
|
||
logger.error(f"【临时修复】手动设置 master_users 失败: {patch_exc}")
|
||
|
||
return config
|
||
except Exception as e:
|
||
logger.critical(f"配置文件解析失败: {e}")
|
||
raise e
|
||
|
||
|
||
def api_ada_load_config(config_path: str) -> APIAdapterConfig:
|
||
"""
|
||
加载API适配器配置文件
|
||
Args:
|
||
config_path: 配置文件路径
|
||
Returns:
|
||
APIAdapterConfig对象
|
||
"""
|
||
# 读取配置文件
|
||
with open(config_path, encoding="utf-8") as f:
|
||
config_data = tomlkit.load(f)
|
||
|
||
config_dict = dict(config_data)
|
||
|
||
try:
|
||
logger.debug("正在解析和验证API适配器配置文件...")
|
||
config = APIAdapterConfig.from_dict(config_dict)
|
||
logger.debug("API适配器配置文件解析和验证完成")
|
||
return config
|
||
except Exception as e:
|
||
logger.critical(f"API适配器配置文件解析失败: {e}")
|
||
raise e
|
||
|
||
|
||
# 获取配置文件路径
|
||
|
||
def initialize_configs_once() -> tuple[Config, APIAdapterConfig]:
|
||
"""
|
||
初始化配置文件，只执行一次。
|
||
"""
|
||
global _CONFIG_INITIALIZED, global_config, model_config
|
||
|
||
if _CONFIG_INITIALIZED and global_config and model_config:
|
||
logger.debug("config.py 初始化已执行，跳过重复运行")
|
||
return global_config, model_config
|
||
|
||
logger.debug(f"MaiCore当前版本: {MMC_VERSION}")
|
||
update_config()
|
||
update_model_config()
|
||
|
||
logger.debug("正在品鉴配置文件...")
|
||
global_config = load_config(config_path=os.path.join(CONFIG_DIR, "bot_config.toml"))
|
||
model_config = api_ada_load_config(config_path=os.path.join(CONFIG_DIR, "model_config.toml"))
|
||
|
||
_CONFIG_INITIALIZED = True
|
||
return global_config, model_config
|
||
|
||
|
||
# 同一进程只执行一次初始化，避免重复生成或覆盖配置
|
||
global_config, model_config = initialize_configs_once()
|
||
|
||
logger.debug("非常的新鲜，非常的美味！") |