🐛 修复config.yaml可能丢失配置的问题

.env.dev

@@ -1,77 +0,0 @@
-SUPERUSERS=[""]
-
-COMMAND_START=[""]
-
-SESSION_RUNNING_EXPRESSION="别急呀,小真寻要宕机了!QAQ"
-
-NICKNAME=["真寻", "小真寻", "绪山真寻", "小寻子"]
-
-SESSION_EXPIRE_TIMEOUT=00:00:30
-
-ALCONNA_USE_COMMAND_START=True
-
-# 全局图片统一使用bytes发送，当真寻与协议端不在同一服务器上时为True
-IMAGE_TO_BYTES = True
-
-# 回复消息时自称
-SELF_NICKNAME="小真寻"
-
-# 官bot appid:bot账号
-QBOT_ID_DATA = '{
-    
-}'
-
-# 数据库配置
-# 示例: "postgres://user:password@127.0.0.1:5432/database"
-# 示例: "mysql://user:password@127.0.0.1:3306/database"
-# 示例: "sqlite:data/db/zhenxun.db"   在data目录下建立db文件夹
-DB_URL = ""
-
-# 系统代理
-# SYSTEM_PROXY = "http://127.0.0.1:7890"
-
-PLATFORM_SUPERUSERS = '
-  {
-    "qq": [""],
-    "dodo": [""]
-  }
-'
-
-DRIVER=~fastapi+~httpx+~websockets
-
-
-# LOG_LEVEL=DEBUG
-# 服务器和端口
-HOST = 127.0.0.1
-PORT = 8080
-
-# kook adapter toekn
-# kaiheila_bots =[{"token": ""}]
-
-# # discode adapter
-# DISCORD_BOTS='
-# [
-#   {
-#     "token": "",
-#     "intent": {
-#       "guild_messages": true,
-#       "direct_messages": true
-#     },
-#     "application_commands": {"*": ["*"]}
-#   }
-# ]
-# '
-# DISCORD_PROXY=''
-
-# # dodo adapter
-# DODO_BOTS='
-# [
-#   {
-#     "client_id": "",
-#     "token": ""
-#   }
-# ]
-# '
-
-# application_commands的{"*": ["*"]}代表将全部应用命令注册为全局应用命令
-# {"admin": ["123", "456"]}则代表将admin命令注册为id是123、456服务器的局部命令，其余命令不注册

zhenxun/configs/utils/__init__.py

@@ -366,24 +366,42 @@ class ConfigsManager:
 
         if not module or not key:
             raise ValueError("add_plugin_config: module和key不能为为空")
+        
+        key_upper = key.upper()
         self.add_module.append(f"{module}:{key}".lower())
-        if module in self._data and (config := self._data[module].configs.get(key)):
+        
-            config.help = help
+        # 检查配置是否已存在
-            config.arg_parser = arg_parser
+        config_exists = module in self._data and key_upper in self._data[module].configs
-            config.type = type
+        
+        # 如果配置已存在，仅更新元数据，保留原始值
+        if config_exists:
+            config = self._data[module].configs[key_upper]
+            # 只更新元数据
+            if help is not None:
+                config.help = help
+            if arg_parser is not None:
+                config.arg_parser = arg_parser
+            if type is not None:
+                config.type = type
+            # 只在强制覆盖模式下才更新值
             if _override:
                 config.value = value
                 config.default_value = default_value
         else:
-            key = key.upper()
+            # 配置不存在，创建新配置
             if not self._data.get(module):
                 self._data[module] = ConfigGroup(module=module)
-            self._data[module].configs[key] = ConfigModel(
+            self._data[module].configs[key_upper] = ConfigModel(
                 value=value,
                 help=help,
                 default_value=default_value,
                 type=type,
             )
+            
+            # 同时更新simple_data
+            if module not in self._simple_data:
+                self._simple_data[module] = {}
+            self._simple_data[module][key_upper] = value
 
     def set_config(
         self,
@@ -486,30 +504,63 @@ class ConfigsManager:
             path: 路径.
             save_simple_data: 同时保存至config.yaml.
         """
-        if save_simple_data:
+        try:
-            with open(self._simple_file, "w", encoding="utf8") as f:
+            if save_simple_data:
-                _yaml.dump(self._simple_data, f)
+                # 使用临时文件进行原子写入
-        path = path or self.file
+                temp_simple_file = self._simple_file.with_suffix(".yaml.tmp")
-        data = {}
+                with open(temp_simple_file, "w", encoding="utf8") as f:
-        for module in self._data:
+                    _yaml.dump(self._simple_data, f)
-            data[module] = {}
+                # 原子替换
-            for config in self._data[module].configs:
+                temp_simple_file.replace(self._simple_file)
-                value = self._data[module].configs[config].dict()
+                
-                del value["type"]
+            path = path or self.file
-                del value["arg_parser"]
+            data = {}
-                data[module][config] = value
+            for module in self._data:
-        with open(path, "w", encoding="utf8") as f:
+                data[module] = {}
-            _yaml.dump(data, f)
+                for config in self._data[module].configs:
+                    value = self._data[module].configs[config].dict()
+                    del value["type"]
+                    del value["arg_parser"]
+                    data[module][config] = value
+                    
+            # 使用临时文件进行原子写入
+            temp_file = Path(str(path) + ".tmp")
+            with open(temp_file, "w", encoding="utf8") as f:
+                _yaml.dump(data, f)
+            # 原子替换
+            temp_file.replace(path)
+        except Exception as e:
+            logger.error(f"保存配置文件失败: {str(e)}")
 
     def reload(self):
         """重新加载配置文件"""
-        if self._simple_file.exists():
+        try:
-            with open(self._simple_file, encoding="utf8") as f:
+            # 先备份当前配置
-                self._simple_data = _yaml.load(f)
+            backup_data = copy.deepcopy(self._data)
-        for key in self._simple_data.keys():
+            
-            for k in self._simple_data[key].keys():
+            if self._simple_file.exists():
-                self._data[key].configs[k].value = self._simple_data[key][k]
+                with open(self._simple_file, encoding="utf8") as f:
-        self.save()
+                    self._simple_data = _yaml.load(f)
+                    
+            # 检查加载的数据是否为None
+            if self._simple_data is None:
+                logger.error("配置文件为空或格式错误，保留原有配置")
+                self._simple_data = {}
+                
+            # 更新配置值
+            for key in self._simple_data.keys():
+                for k in self._simple_data[key].keys():
+                    if key in self._data and k in self._data[key].configs:
+                        self._data[key].configs[k].value = self._simple_data[key][k]
+            
+            # 保存更新后的配置
+            self.save()
+        except ScannerError as e:
+            logger.error(f"配置文件解析失败: {str(e)}，保留现有配置")
+        except Exception as e:
+            logger.error(f"重新加载配置失败: {str(e)}")
+            # 发生错误时恢复到备份配置
+            self._data = backup_data
 
     def load_data(self):
         """加载数据