feat(config): centralize set_key usage through update_setting method

app/api/endpoints/system.py

@@ -4,7 +4,6 @@ from datetime import datetime
 from typing import Union, Any
 
 import tailer
-from dotenv import set_key
 from fastapi import APIRouter, HTTPException, Depends, Response
 from fastapi.responses import StreamingResponse
 
@@ -23,6 +22,7 @@ from app.helper.rule import RuleHelper
 from app.helper.sites import SitesHelper
 from app.monitor import Monitor
 from app.scheduler import Scheduler
+from app.schemas.types import SystemConfigKey
 from app.utils.http import RequestUtils
 from app.utils.system import SystemUtils
 from version import APP_VERSION
@@ -112,19 +112,28 @@ def set_env_setting(env: dict,
     """
     更新系统环境变量（仅管理员）
     """
-    for k, v in env.items():
+    result = settings.update_settings(env=env)
-        if k == "undefined":
+    # 统计成功和失败的结果
-            continue
+    success_updates = {k: v for k, v in result.items() if v[0]}
-        if hasattr(settings, k):
+    failed_updates = {k: v for k, v in result.items() if not v[0]}
-            if v == "None":
+
-                v = None
+    if failed_updates:
-            setattr(settings, k, v)
+        return schemas.Response(
-            if v is None:
+            success=False,
-                v = ''
+            message="部分配置项更新失败",
-            else:
+            data={
-                v = str(v)
+                "success_updates": success_updates,
-            set_key(SystemUtils.get_env_path(), k, v)
+                "failed_updates": failed_updates
-    return schemas.Response(success=True)
+            }
+        )
+
+    return schemas.Response(
+        success=True,
+        message="所有配置项更新成功",
+        data={
+            "success_updates": success_updates
+        }
+    )
 
 
 @router.get("/progress/{process_type}", summary="实时进度")
@@ -173,17 +182,13 @@ def set_setting(key: str, value: Union[list, dict, bool, int, str] = None,
     更新系统设置（仅管理员）
     """
     if hasattr(settings, key):
-        if value == "None":
+        success, message = settings.update_setting(key=key, value=value)
-            value = None
+        return schemas.Response(success=success, message=message)
-        setattr(settings, key, value)
+    elif key in {item.value for item in SystemConfigKey}:
-        if value is None:
-            value = ''
-        else:
-            value = str(value)
-        set_key(SystemUtils.get_env_path(), key, value)
-    else:
         SystemConfigOper().set(key, value)
         return schemas.Response(success=True)
+    else:
+        return schemas.Response(success=False, message=f"配置项 '{key}' 不存在")
 
 
 @router.get("/message", summary="实时消息")

app/core/config.py

@@ -1,9 +1,10 @@
 import os
+import re
 import secrets
 import sys
 import threading
 from pathlib import Path
-from typing import Optional, List, Any, Type, Tuple
+from typing import Optional, List, Any, Type, Tuple, Dict
 
 from dotenv import set_key
 from pydantic import BaseSettings, validator, BaseModel
@@ -228,13 +229,14 @@ class Settings(BaseSettings, ConfigModel):
         return v
 
     @staticmethod
-    def generic_type_converter(value: Any, expected_type: Type, default: Any, field_name: str) -> Tuple[Any, bool]:
+    def generic_type_converter(value: Any, original_value: Any, expected_type: Type, default: Any, field_name: str,
+                               raise_exception: bool = False) -> Tuple[Any, bool]:
+        """
+        通用类型转换函数，根据预期类型转换值。如果转换失败，返回默认值
         """
-        通用类型转换函数，根据预期类型转换值。如果转换失败，返回默认值        """
         if value is None:
             return default, False
 
-        original_value = value
         if isinstance(value, str):
             value = value.strip()
 
@@ -266,6 +268,10 @@ class Settings(BaseSettings, ConfigModel):
                     converted = float(value)
                     return converted, value != original_value
             elif expected_type is str:
+                # 清理 value 中所有空白字符的字段
+                fields_not_keep_spaces = {"AUTO_DOWNLOAD_USER", "REPO_GITHUB_TOKEN", "PLUGIN_MARKET"}
+                if field_name in fields_not_keep_spaces:
+                    value = re.sub(r"\s+", "", value)
                 return value, value != original_value
             # # 后续考虑支持 list 类型的处理
             # elif expected_type is list:
@@ -277,25 +283,71 @@ class Settings(BaseSettings, ConfigModel):
             # 可根据需要添加更多类型处理
             else:
                 return value, False
-        except (ValueError, TypeError):
+        except (ValueError, TypeError) as e:
+            if raise_exception:
+                raise ValueError(f"配置项 '{field_name}' 的值 '{value}' 无法转换成正确的类型") from e
+            logger.error(
+                f"配置项 '{field_name}' 的值 '{value}' 无法转换成正确的类型，使用默认值 '{default}'，错误信息: {e}")
             return default, True
 
     @validator('*', pre=True, always=True)
     def generic_type_validator(cls, value: Any, field):
         """
-        通用校验器
+        通用校验器，尝试将配置值转换为期望的类型
         """
-        converted_value, needs_update = cls.generic_type_converter(value, field.type_, field.default, field.name)
+        converted_value, needs_update = cls.generic_type_converter(value, value, field.type_, field.default,
+                                                                   field.name)
         if needs_update:
-            logger.error(f"字段 '{field.name}' 的值 '{value}' 无效，已使用 '{converted_value}' 进行替换")
+            cls.update_env_config(field, value, converted_value)
-            if field.name in os.environ:
-                logger.warning(f"字段 '{field.name}' 已存在于环境变量中，请手动修改")
-            else:
-                set_key(SystemUtils.get_env_path(), field.name,
-                        str(converted_value) if converted_value is not None else "")
-                logger.info(f"字段 '{field.name}' 已由应用修改并写入到 app.env 中")
         return converted_value
 
+    @staticmethod
+    def update_env_config(field: Any, value: Any, converted_value: Any) -> Tuple[bool, str]:
+        """
+        更新 env 配置
+        """
+        is_converted = value is not None and value != converted_value
+        if is_converted:
+            logger.warning(f"配置项 '{field.name}' 的值 '{value}' 无效，已替换为 '{converted_value}'")
+
+        if field.name in os.environ:
+            if is_converted:
+                message = f"配置项 '{field.name}' 已在环境变量中设置，请手动更新以保持一致性"
+                logger.warning(message)
+                return False, message
+            return True, ""
+        else:
+            set_key(SystemUtils.get_env_path(), field.name, str(converted_value) if converted_value is not None else "")
+            if is_converted:
+                logger.info(f"配置项 '{field.name}' 已自动修正并写入到 'app.env' 文件中")
+            return True, ""
+
+    def update_setting(self, key: str, value: Any) -> Tuple[bool, str]:
+        """
+        更新单个配置项
+        """
+        if not hasattr(self, key):
+            return False, f"配置项 '{key}' 不存在"
+
+        try:
+            field = self.__fields__[key]
+            converted_value, _ = self.generic_type_converter(value, getattr(self, key), field.type_,
+                                                             field.default, key)
+            # 如果没有抛出异常，则统一使用 converted_value 进行更新
+            setattr(self, key, converted_value)
+            return self.update_env_config(field, value, converted_value)
+        except Exception as e:
+            return False, str(e)
+
+    def update_settings(self, env: Dict[str, Any]) -> Dict[str, Tuple[bool, str]]:
+        """
+        更新多个配置项
+        """
+        results = {}
+        for k, v in env.items():
+            results[k] = self.update_setting(k, v)
+        return results
+
     @property
     def VERSION_FLAG(self) -> str:
         """