From 9d71c9b61eff56a129b382d3fed8cdb0046f37fe Mon Sep 17 00:00:00 2001 From: InfinityPacer <160988576+InfinityPacer@users.noreply.github.com> Date: Sat, 5 Oct 2024 03:14:16 +0800 Subject: [PATCH] feat(config): centralize set_key usage through update_setting method --- app/api/endpoints/system.py | 53 +++++++++++++----------- app/core/config.py | 80 ++++++++++++++++++++++++++++++------- 2 files changed, 95 insertions(+), 38 deletions(-) diff --git a/app/api/endpoints/system.py b/app/api/endpoints/system.py index e21ecb44..7704c788 100644 --- a/app/api/endpoints/system.py +++ b/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(): - if k == "undefined": - continue - if hasattr(settings, k): - if v == "None": - v = None - setattr(settings, k, v) - if v is None: - v = '' - else: - v = str(v) - set_key(SystemUtils.get_env_path(), k, v) - return schemas.Response(success=True) + result = settings.update_settings(env=env) + # 统计成功和失败的结果 + success_updates = {k: v for k, v in result.items() if v[0]} + failed_updates = {k: v for k, v in result.items() if not v[0]} + + if failed_updates: + return schemas.Response( + success=False, + message="部分配置项更新失败", + data={ + "success_updates": success_updates, + "failed_updates": failed_updates + } + ) + + 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": - value = None - setattr(settings, key, value) - if value is None: - value = '' - else: - value = str(value) - set_key(SystemUtils.get_env_path(), key, value) - else: + success, message = settings.update_setting(key=key, value=value) + return schemas.Response(success=success, message=message) + elif key in {item.value for item in SystemConfigKey}: SystemConfigOper().set(key, value) - return schemas.Response(success=True) + return schemas.Response(success=True) + else: + return schemas.Response(success=False, message=f"配置项 '{key}' 不存在") @router.get("/message", summary="实时消息") diff --git a/app/core/config.py b/app/core/config.py index 9c70740d..2adaf8d1 100644 --- a/app/core/config.py +++ b/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}' 进行替换") - 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 中") + cls.update_env_config(field, value, converted_value) 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: """