update 模块解耦

src/application/application.py

@@ -18,7 +18,15 @@ from ..presentation import (
     create_flask_app,
 )
 from ..repositories import LogRepository, UserRepository
-from ..services import AuthenticationService, LogService, ProviderService, ProxyService, UserService
+from ..services import (
+    AuthenticationService,
+    LogService,
+    ModelDiscoveryService,
+    ProviderService,
+    ProxyService,
+    SettingsService,
+    UserService,
+)
 from ..utils import normalize_ip
 from ..utils.database import create_connection_factory
 
@@ -39,7 +47,7 @@ class Application:
         self._setup_controllers()
         self._setup_request_access_logging()
 
-        self._logger.info("Application initialized successfully")
+        self._logger.info('Application initialized successfully')
 
     def _setup_config(self) -> None:
         """初始化配置管理器。"""
@@ -54,18 +62,18 @@ class Application:
         level = getattr(logging, log_level.upper(), logging.INFO)
 
         formatter = logging.Formatter(
-            "%(asctime)s|%(name)s|%(filename)s:%(lineno)d|%(levelname)s|%(message)s",
-            datefmt="%Y-%m-%d %H:%M:%S",
+            '%(asctime)s|%(name)s|%(filename)s:%(lineno)d|%(levelname)s|%(message)s',
+            datefmt='%Y-%m-%d %H:%M:%S',
         )
 
-        logger = logging.getLogger("app")
+        logger = logging.getLogger('app')
         logger.handlers.clear()
 
         app_log_handler = RotatingFileHandler(
-            log_path / "app.log",
+            log_path / 'app.log',
             maxBytes=10 * 1024 * 1024,
             backupCount=3,
-            encoding="utf-8",
+            encoding='utf-8',
         )
         app_log_handler.setFormatter(formatter)
         app_log_handler.setLevel(level)
@@ -77,18 +85,18 @@ class Application:
         logger.addHandler(console_handler)
         logger.setLevel(level)
 
-        access_logger = logging.getLogger("access")
+        access_logger = logging.getLogger('access')
         access_logger.handlers.clear()
         access_handler = RotatingFileHandler(
-            log_path / "access.log",
+            log_path / 'access.log',
             maxBytes=10 * 1024 * 1024,
             backupCount=3,
-            encoding="utf-8",
+            encoding='utf-8',
         )
         access_handler.setFormatter(
             logging.Formatter(
-                "%(asctime)s|%(name)s|%(filename)s:%(lineno)d|%(levelname)s|%(message)s",
-                datefmt="%Y-%m-%d %H:%M:%S",
+                '%(asctime)s|%(name)s|%(filename)s:%(lineno)d|%(levelname)s|%(message)s',
+                datefmt='%Y-%m-%d %H:%M:%S',
             )
         )
         access_handler.setLevel(level)
@@ -100,28 +108,28 @@ class Application:
         self._flask_app.access_logger = access_logger
 
     def _setup_request_access_logging(self) -> None:
-        """注册请求访问日志钩子，仅记录 IP 与 URL。"""
+        """注册请求访问日志钩子，仅记录 IP、URL 与模型。"""
 
         @self._flask_app.before_request
         def log_request_access() -> None:
-            client_ip = normalize_ip(request.remote_addr) or "-"
+            client_ip = normalize_ip(request.remote_addr) or '-'
             requested_url = request.url
             model = None
             if (request.content_length or 0) > 0 and request.is_json:
                 payload = request.get_json(silent=True)
                 if isinstance(payload, dict):
-                    model = payload.get("model")
+                    model = payload.get('model')
 
             username = None
             user = self._user_service.get_user_by_ip(client_ip)
             if user:
-                username = str(user.get("username") or "")
+                username = str(user.get('username') or '')
 
-            log_message = f"ip={client_ip} url={requested_url}"
+            log_message = f'ip={client_ip} url={requested_url}'
             if username:
-                log_message = f"{log_message} username={username}"
+                log_message = f'{log_message} username={username}'
             if model is not None:
-                log_message = f"{log_message} model={model}"
+                log_message = f'{log_message} model={model}'
             self._flask_app.access_logger.info(log_message)
 
     def _setup_context(self) -> None:
@@ -152,22 +160,34 @@ class Application:
         self._user_service = user_service
         proxy_service = ProxyService(self._ctx)
         log_service = LogService(self._ctx, self._log_repository)
-        provider_service = ProviderService(self._ctx, self._config_path, self.reload_providers)
+        provider_service = ProviderService(self._ctx, self.reload_providers)
+        model_discovery_service = ModelDiscoveryService(self._ctx)
+        settings_service = SettingsService(self._ctx)
 
         self._auth_controller = AuthenticationController(self._ctx, auth_service)
         self._user_controller = UserController(self._ctx, user_service, auth_service)
-        self._provider_controller = ProviderController(self._ctx, provider_service, auth_service)
+        self._provider_controller = ProviderController(
+            self._ctx,
+            provider_service,
+            model_discovery_service,
+            settings_service,
+            auth_service,
+        )
         self._proxy_controller = ProxyController(
-            self._ctx, proxy_service, user_service, log_service, self._provider_manager
+            self._ctx,
+            proxy_service,
+            user_service,
+            log_service,
+            self._provider_manager,
         )
         self._web_controller = WebController(self._ctx, log_service, auth_service)
 
-        self._logger.info("All controllers initialized successfully")
+        self._logger.info('All controllers initialized successfully')
 
     def reload_providers(self) -> None:
         self._config_manager.reload()
         config_dict = self._config_manager.get_raw_config()
-        providers_config = config_dict.get("providers", [])
+        providers_config = config_dict.get('providers', [])
         self._provider_manager.load_providers(providers_config)
 
     def run(self) -> None:
@@ -178,5 +198,5 @@ class Application:
         from gevent.pywsgi import WSGIServer
 
         server = WSGIServer((host, port), self._flask_app)
-        self._logger.info(f"Starting LLM Proxy on {host}:{port}...")
+        self._logger.info('Starting LLM Proxy on %s:%s...', host, port)
         server.serve_forever()

src/config/config_manager.py

@@ -2,6 +2,9 @@
 # -*- coding: utf-8 -*-
 """配置访问层。"""
 
+import os
+import tempfile
+from copy import deepcopy
 from pathlib import Path
 from typing import Any, Dict, Optional, Union
 
@@ -33,7 +36,7 @@ class ConfigManager:
 
     def get_admin_config(self) -> Optional[Dict[str, str]]:
         admin = self.get('admin')
-        return admin if isinstance(admin, dict) else None
+        return dict(admin) if isinstance(admin, dict) else None
 
     def is_auth_enabled(self) -> bool:
         admin = self.get_admin_config()
@@ -59,7 +62,14 @@ class ConfigManager:
         return self.get('logging.level', 'INFO').upper()
 
     def get_raw_config(self) -> Dict[str, Any]:
-        return self._config.copy()
+        return deepcopy(self._config)
+
+    def write_raw_config(self, config: Dict[str, Any]) -> None:
+        if not isinstance(config, dict):
+            raise ValueError('Configuration file must contain a top-level mapping')
+
+        self._write_config(config)
+        self._config = deepcopy(config)
 
     def reload(self) -> None:
         self._config = self._load_config(self._config_path)
@@ -71,4 +81,29 @@ class ConfigManager:
             raise FileNotFoundError(f'Configuration file not found: {path}')
 
         with open(path, 'r', encoding='utf-8') as file:
-            return yaml.safe_load(file) or {}
+            data = yaml.safe_load(file)
+
+        if data is None:
+            return {}
+        if not isinstance(data, dict):
+            raise ValueError('Configuration file must contain a top-level mapping')
+        return data
+
+    def _write_config(self, config: Dict[str, Any]) -> None:
+        temp_file_path: Optional[str] = None
+        try:
+            with tempfile.NamedTemporaryFile(
+                'w',
+                encoding='utf-8',
+                dir=self._config_path.parent,
+                delete=False,
+            ) as temp_file:
+                yaml.safe_dump(config, temp_file, allow_unicode=True, sort_keys=False)
+                temp_file.flush()
+                os.fsync(temp_file.fileno())
+                temp_file_path = temp_file.name
+
+            os.replace(temp_file_path, self._config_path)
+        finally:
+            if temp_file_path and os.path.exists(temp_file_path):
+                os.remove(temp_file_path)

src/config/provider_config.py

@@ -0,0 +1,184 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""Provider 配置的规范化与校验。"""
+
+from __future__ import annotations
+
+from typing import Any, Dict, List, Mapping, Optional, Sequence
+
+from ..utils.net import normalize_proxy_url
+
+DEFAULT_PROVIDER_TIMEOUT_SECONDS = 300
+DEFAULT_PROVIDER_MAX_RETRIES = 3
+DEFAULT_PROVIDER_VERIFY_SSL = False
+
+
+def clean_optional_string(value: Any) -> Optional[str]:
+    if value is None:
+        return None
+    text = str(value).strip()
+    return text or None
+
+
+def parse_optional_positive_int(
+    value: Any,
+    *,
+    default: Optional[int] = None,
+    error_message: str = "Expected a positive integer",
+) -> Optional[int]:
+    if value is None:
+        return default
+    if isinstance(value, str) and not value.strip():
+        return default
+
+    try:
+        parsed = int(value)
+    except (TypeError, ValueError) as exc:
+        raise ValueError(error_message) from exc
+
+    if parsed <= 0:
+        raise ValueError(error_message)
+    return parsed
+
+
+def parse_optional_bool(
+    value: Any,
+    *,
+    default: Optional[bool] = None,
+    error_message: str = "Expected a boolean value",
+) -> Optional[bool]:
+    if value is None:
+        return default
+
+    if isinstance(value, str):
+        lowered = value.strip().lower()
+        if lowered == "":
+            return default
+        if lowered in {"true", "1", "yes", "on"}:
+            return True
+        if lowered in {"false", "0", "no", "off"}:
+            return False
+        raise ValueError(error_message)
+
+    if isinstance(value, bool):
+        return value
+    if isinstance(value, (int, float)):
+        return bool(value)
+
+    raise ValueError(error_message)
+
+
+def normalize_model_list(value: Any) -> List[str]:
+    models: List[str] = []
+    if value is None:
+        return models
+
+    if isinstance(value, str):
+        raw_items = value.replace(",", "\n").splitlines()
+    elif isinstance(value, list):
+        raw_items = value
+    else:
+        raise ValueError("model_list must be a list or newline-separated string")
+
+    seen: set[str] = set()
+    for item in raw_items:
+        model = str(item).strip()
+        if not model or model in seen:
+            continue
+        seen.add(model)
+        models.append(model)
+    return models
+
+
+def normalize_provider_payload(payload: Mapping[str, Any]) -> Dict[str, Any]:
+    if not isinstance(payload, Mapping):
+        raise ValueError("Provider payload must be an object")
+
+    name = clean_optional_string(payload.get("name"))
+    api = clean_optional_string(payload.get("api"))
+    if name is None:
+        raise ValueError("Provider name is required")
+    if api is None:
+        raise ValueError("Provider api is required")
+
+    normalized: Dict[str, Any] = {
+        "name": name,
+        "api": api,
+    }
+
+    api_key = clean_optional_string(payload.get("api_key"))
+    if api_key is not None:
+        normalized["api_key"] = api_key
+
+    proxy = normalize_proxy_url(payload.get("proxy"))
+    if proxy is not None:
+        normalized["proxy"] = proxy
+
+    timeout_seconds = parse_optional_positive_int(payload.get("timeout_seconds"))
+    if timeout_seconds is not None:
+        normalized["timeout_seconds"] = timeout_seconds
+
+    max_retries = parse_optional_positive_int(payload.get("max_retries"))
+    if max_retries is not None:
+        normalized["max_retries"] = max_retries
+
+    verify_ssl = parse_optional_bool(payload.get("verify_ssl"))
+    if verify_ssl is not None:
+        normalized["verify_ssl"] = verify_ssl
+
+    model_list = normalize_model_list(payload.get("model_list"))
+    if model_list:
+        normalized["model_list"] = model_list
+
+    hook = clean_optional_string(payload.get("hook"))
+    if hook is not None:
+        normalized["hook"] = hook
+
+    return normalized
+
+
+def normalize_runtime_provider_config(config: Mapping[str, Any]) -> Dict[str, Any]:
+    if not isinstance(config, Mapping):
+        raise ValueError("Provider config must be an object")
+
+    name = clean_optional_string(config.get("name"))
+    api = clean_optional_string(config.get("api"))
+    if name is None:
+        raise ValueError("Provider name is required")
+    if api is None:
+        raise ValueError("Provider api is required")
+
+    return {
+        "name": name,
+        "api": api,
+        "api_key": clean_optional_string(config.get("api_key")) or "",
+        "model_list": normalize_model_list(config.get("model_list")),
+        "proxy": normalize_proxy_url(config.get("proxy")),
+        "timeout_seconds": parse_optional_positive_int(
+            config.get("timeout_seconds"),
+            default=DEFAULT_PROVIDER_TIMEOUT_SECONDS,
+        ),
+        "max_retries": parse_optional_positive_int(
+            config.get("max_retries"),
+            default=DEFAULT_PROVIDER_MAX_RETRIES,
+        ),
+        "verify_ssl": parse_optional_bool(
+            config.get("verify_ssl"),
+            default=DEFAULT_PROVIDER_VERIFY_SSL,
+        ),
+        "hook": clean_optional_string(config.get("hook")),
+    }
+
+
+def validate_provider_definitions(providers: Sequence[Mapping[str, Any]]) -> None:
+    seen_names: set[str] = set()
+
+    for index, provider in enumerate(providers):
+        if not isinstance(provider, Mapping):
+            raise ValueError(f"Provider entry at index {index} must be an object")
+
+        normalized = normalize_runtime_provider_config(provider)
+        name = normalized["name"]
+        if name in seen_names:
+            raise ValueError(f"Duplicate provider name detected: {name}")
+        seen_names.add(name)

src/config/provider_manager.py

@@ -1,6 +1,6 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
-"""Provider 管理器：负责加载配置与可选钩子模块。"""
+"""Provider 管理器：负责加载配置、构建运行时对象并维护模型映射。"""
 
 import hashlib
 import importlib.util
@@ -12,14 +12,13 @@ from typing import Any, Dict, List, Optional
 from ..application.app_context import AppContext, Logger
 from ..external import LLMProvider
 from ..hooks import HookModule
-from ..utils.net import normalize_proxy_url
+from .provider_config import normalize_runtime_provider_config
 
 
 class ProviderManager:
-    """管理模型与 provider 的映射关系。"""
+    """管理模型到 provider 的映射关系。"""
 
     def __init__(self, ctx: AppContext):
-        self._ctx = ctx
         self._base_dir = ctx.root_path.resolve()
         self._logger: Logger = ctx.logger
         self._provider_by_model: Dict[str, LLMProvider] = {}
@@ -37,38 +36,27 @@ class ProviderManager:
 
     def _load_provider(self, config: Dict[str, Any]) -> None:
         """加载单个 provider，并注册其模型映射。"""
-        name = config.get('name')
-        api = config.get('api', '')
-        api_key = config.get('api_key', '')
-        model_list = config.get('model_list', [])
-        proxy = normalize_proxy_url(config.get('proxy'))
-        timeout_seconds = self._to_positive_int(config.get('timeout_seconds', 300), default=300)
-        max_retries = self._to_positive_int(config.get('max_retries', 3), default=3)
-        verify_ssl = self._to_bool(config.get('verify_ssl', False), default=False)
+        normalized = normalize_runtime_provider_config(config)
+        name = normalized['name']
+        model_list = normalized['model_list']
 
         if name in self._provider_names:
-            raise ValueError(f"Duplicate provider name detected: {name}")
-
-        if not api:
-            self._logger.warning(f"Provider '{name}' skipped: api is empty")
-            return
+            raise ValueError(f'Duplicate provider name detected: {name}')
 
         if not model_list:
-            self._logger.warning(f"Provider '{name}' skipped: model_list is empty")
+            self._logger.warning("Provider '%s' skipped: model_list is empty", name)
             return
 
-        hook = self._load_hook(config.get('hook'))
-
         provider = LLMProvider(
             name=name,
-            api=api,
-            api_key=api_key,
+            api=normalized['api'],
+            api_key=normalized['api_key'],
             model_list=model_list,
-            proxy=proxy,
-            timeout_seconds=timeout_seconds,
-            max_retries=max_retries,
-            verify_ssl=verify_ssl,
-            hook=hook,
+            proxy=normalized['proxy'],
+            timeout_seconds=normalized['timeout_seconds'],
+            max_retries=normalized['max_retries'],
+            verify_ssl=normalized['verify_ssl'],
+            hook=self._load_hook(normalized['hook']),
         )
 
         for model in model_list:
@@ -78,25 +66,24 @@ class ProviderManager:
             self._provider_by_model[key] = provider
 
         self._provider_names.add(name)
-        self._logger.info(f'Loaded provider: {name}, models: {model_list}')
+        self._logger.info('Loaded provider: %s, models: %s', name, model_list)
 
     def _load_hook(self, hook_path: Optional[str]) -> Optional[HookModule]:
-        """按路径加载钩子模块，并做缓存复用。"""
+        """按路径加载 hook 模块，并做缓存复用。"""
         if not hook_path:
             return None
 
         if Path(hook_path).is_absolute():
             hook_file = Path(hook_path).resolve()
         else:
-            hooks_dir = self._base_dir / 'hooks'
-            hook_file = (hooks_dir / hook_path).resolve()
+            hook_file = (self._base_dir / 'hooks' / hook_path).resolve()
 
         cache_key = str(hook_file)
         if cache_key in self._hook_cache:
             return self._hook_cache[cache_key]
 
         if not hook_file.exists():
-            self._logger.warning(f'Hook file not found: {hook_file}')
+            self._logger.warning('Hook file not found: %s', hook_file)
             self._hook_cache[cache_key] = None
             return None
 
@@ -117,15 +104,16 @@ class ProviderManager:
                 spec.loader.exec_module(module)
                 hook_class = getattr(module, 'Hook', None)
                 if hook_class is None or not inspect.isclass(hook_class):
-                    self._logger.error(f'Hook file must export a class named Hook: {hook_file}')
+                    self._logger.error('Hook file must export a class named Hook: %s', hook_file)
                     self._hook_cache[cache_key] = None
                     return None
+
                 hook_instance = hook_class()
                 self._hook_cache[cache_key] = hook_instance
-                self._logger.info(f'Hook loaded successfully: {hook_file}')
+                self._logger.info('Hook loaded successfully: %s', hook_file)
                 return hook_instance
         except Exception as exc:
-            self._logger.error(f'Failed to load hook {hook_file}: {exc}')
+            self._logger.error('Failed to load hook %s: %s', hook_file, exc)
         finally:
             if path_inserted:
                 try:
@@ -141,25 +129,3 @@ class ProviderManager:
 
     def get_all_models(self) -> List[str]:
         return sorted(self._provider_by_model.keys())
-
-    @staticmethod
-    def _to_positive_int(value: Any, default: int) -> int:
-        try:
-            parsed = int(value)
-            return parsed if parsed > 0 else default
-        except (TypeError, ValueError):
-            return default
-
-    @staticmethod
-    def _to_bool(value: Any, default: bool) -> bool:
-        if isinstance(value, bool):
-            return value
-        if isinstance(value, str):
-            lowered = value.strip().lower()
-            if lowered in {'true', '1', 'yes', 'on'}:
-                return True
-            if lowered in {'false', '0', 'no', 'off'}:
-                return False
-        if isinstance(value, (int, float)):
-            return bool(value)
-        return default

src/external/llm_provider.py

@@ -1,6 +1,6 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
-"""LLM Provider 配置与请求阶段钩子封装。"""
+"""LLM provider 运行时对象。"""
 
 from typing import Any, Dict, Optional
 
@@ -8,7 +8,7 @@ from ..hooks import HookContext, HookModule
 
 
 class LLMProvider:
-    """封装 provider 配置与请求阶段钩子调用。"""
+    """封装 provider 配置与请求阶段 hook 调用。"""
 
     def __init__(
         self,
@@ -25,7 +25,7 @@ class LLMProvider:
         self.name = name
         self.api = api
         self.api_key = api_key
-        self.model_list = model_list or []
+        self.model_list = list(model_list or [])
         self.proxy = proxy
         self.timeout_seconds = timeout_seconds
         self.max_retries = max_retries
@@ -33,11 +33,11 @@ class LLMProvider:
         self.hook = hook
 
     def apply_header_hook(self, ctx: HookContext, headers: Dict[str, str]) -> Dict[str, str]:
-        if self.hook and hasattr(self.hook, "header_hook"):
+        if self.hook and hasattr(self.hook, 'header_hook'):
             return self.hook.header_hook(ctx, headers)
         return headers
 
     def apply_input_body_hook(self, ctx: HookContext, body: Dict[str, Any]) -> Dict[str, Any]:
-        if self.hook and hasattr(self.hook, "input_body_hook"):
+        if self.hook and hasattr(self.hook, 'input_body_hook'):
             return self.hook.input_body_hook(ctx, body)
-        return body
+        return body

src/presentation/provider_controller.py

@@ -5,122 +5,130 @@
 from flask import Response, jsonify, request
 
 from ..application.app_context import AppContext
-from ..services import AuthenticationService, ProviderService
+from ..services import (
+    AuthenticationService,
+    ModelDiscoveryService,
+    ProviderService,
+    SettingsService,
+)
 from .decorators import require_authentication
 
 
 class ProviderController:
-    """处理 provider 配置的增删改查与模型拉取。"""
+    """处理 provider 配置的增删改查、模型拉取与配置开关更新。"""
 
     def __init__(
         self,
         ctx: AppContext,
         provider_service: ProviderService,
+        model_discovery_service: ModelDiscoveryService,
+        settings_service: SettingsService,
         auth_service: AuthenticationService,
     ):
-        self._ctx = ctx
         self._app = ctx.flask_app
         self._logger = ctx.logger
         self._provider_service = provider_service
+        self._model_discovery_service = model_discovery_service
+        self._settings_service = settings_service
         self._auth_service = auth_service
         self._register_routes()
 
     def _register_routes(self) -> None:
         auth = require_authentication(self._auth_service)
 
-        self._app.route("/api/providers", methods=["GET"])(auth(self.get_providers))
-        self._app.route("/api/providers", methods=["POST"])(auth(self.create_provider))
-        self._app.route("/api/providers/fetch-models", methods=["GET"])(auth(self.fetch_models))
-        self._app.route("/api/providers/chat-whitelist", methods=["PUT"])(auth(self.update_chat_whitelist))
-        self._app.route("/api/providers/<string:name>", methods=["GET"])(auth(self.get_provider))
-        self._app.route("/api/providers/<string:name>", methods=["PUT"])(auth(self.update_provider))
-        self._app.route("/api/providers/<string:name>", methods=["DELETE"])(auth(self.delete_provider))
+        self._app.route('/api/providers', methods=['GET'])(auth(self.get_providers))
+        self._app.route('/api/providers', methods=['POST'])(auth(self.create_provider))
+        self._app.route('/api/providers/fetch-models', methods=['GET'])(auth(self.fetch_models))
+        self._app.route('/api/providers/chat-whitelist', methods=['PUT'])(auth(self.update_chat_whitelist))
+        self._app.route('/api/providers/<string:name>', methods=['GET'])(auth(self.get_provider))
+        self._app.route('/api/providers/<string:name>', methods=['PUT'])(auth(self.update_provider))
+        self._app.route('/api/providers/<string:name>', methods=['DELETE'])(auth(self.delete_provider))
 
     def get_providers(self) -> Response:
         try:
             return jsonify(self._provider_service.list_providers())
         except Exception as exc:
-            self._logger.error(f"Error getting providers: {exc}")
-            return jsonify({"error": str(exc)}), 500
+            self._logger.error('Error getting providers: %s', exc)
+            return jsonify({'error': str(exc)}), 500
 
     def get_provider(self, name: str) -> Response:
         try:
             provider = self._provider_service.get_provider(name)
             if provider is None:
-                return jsonify({"error": "Provider not found"}), 404
+                return jsonify({'error': 'Provider not found'}), 404
             return jsonify(provider)
         except Exception as exc:
-            self._logger.error(f"Error getting provider: {exc}")
-            return jsonify({"error": str(exc)}), 500
+            self._logger.error('Error getting provider: %s', exc)
+            return jsonify({'error': str(exc)}), 500
 
     def create_provider(self) -> Response:
         try:
             payload = request.get_json(silent=True) or {}
             provider = self._provider_service.create_provider(payload)
-            self._logger.info(f"Provider created: {provider.get('name')}")
+            self._logger.info('Provider created: %s', provider.get('name'))
             return jsonify(provider), 201
         except ValueError as exc:
-            return jsonify({"error": str(exc)}), 400
+            return jsonify({'error': str(exc)}), 400
         except Exception as exc:
-            self._logger.error(f"Error creating provider: {exc}")
-            return jsonify({"error": str(exc)}), 500
+            self._logger.error('Error creating provider: %s', exc)
+            return jsonify({'error': str(exc)}), 500
 
     def update_provider(self, name: str) -> Response:
         try:
             payload = request.get_json(silent=True) or {}
             provider = self._provider_service.update_provider(name, payload)
-            self._logger.info(f"Provider updated: {name} -> {provider.get('name')}")
+            self._logger.info('Provider updated: %s -> %s', name, provider.get('name'))
             return jsonify(provider)
         except ValueError as exc:
             message = str(exc)
-            status_code = 404 if "not found" in message.lower() else 400
-            return jsonify({"error": message}), status_code
+            status_code = 404 if 'not found' in message.lower() else 400
+            return jsonify({'error': message}), status_code
         except Exception as exc:
-            self._logger.error(f"Error updating provider: {exc}")
-            return jsonify({"error": str(exc)}), 500
+            self._logger.error('Error updating provider: %s', exc)
+            return jsonify({'error': str(exc)}), 500
 
     def delete_provider(self, name: str) -> Response:
         try:
             self._provider_service.delete_provider(name)
-            self._logger.info(f"Provider deleted: {name}")
-            return jsonify({"message": "Provider deleted successfully"})
+            self._logger.info('Provider deleted: %s', name)
+            return jsonify({'message': 'Provider deleted successfully'})
         except ValueError as exc:
             message = str(exc)
-            status_code = 404 if "not found" in message.lower() else 400
-            return jsonify({"error": message}), status_code
+            status_code = 404 if 'not found' in message.lower() else 400
+            return jsonify({'error': message}), status_code
         except Exception as exc:
-            self._logger.error(f"Error deleting provider: {exc}")
-            return jsonify({"error": str(exc)}), 500
+            self._logger.error('Error deleting provider: %s', exc)
+            return jsonify({'error': str(exc)}), 500
 
     def fetch_models(self) -> Response:
         try:
-            result = self._provider_service.fetch_models_preview(
-                api=request.args.get("api", ""),
-                api_key=request.args.get("api_key"),
-                proxy=request.args.get("proxy"),
-                timeout_seconds=request.args.get("timeout_seconds"),
-                verify_ssl=request.args.get("verify_ssl"),
+            result = self._model_discovery_service.fetch_models_preview(
+                api=request.args.get('api', ''),
+                api_key=request.args.get('api_key'),
+                proxy=request.args.get('proxy'),
+                timeout_seconds=request.args.get('timeout_seconds'),
+                verify_ssl=request.args.get('verify_ssl'),
             )
             self._logger.info(
-                "Provider models preview fetched: api=%s fetched=%s",
-                request.args.get("api", ""),
-                len(result["fetched_models"]),
+                'Provider models preview fetched: api=%s fetched=%s',
+                request.args.get('api', ''),
+                len(result['fetched_models']),
             )
             return jsonify(result)
         except ValueError as exc:
-            return jsonify({"error": str(exc)}), 400
+            return jsonify({'error': str(exc)}), 400
         except Exception as exc:
-            self._logger.error(f"Error fetching provider models: {exc}")
-            return jsonify({"error": str(exc)}), 500
+            self._logger.error('Error fetching provider models: %s', exc)
+            return jsonify({'error': str(exc)}), 500
 
     def update_chat_whitelist(self) -> Response:
         try:
             payload = request.get_json(silent=True) or {}
-            enabled = self._provider_service.update_chat_whitelist_enabled(payload.get("enabled"))
-            self._logger.info("Chat whitelist updated: enabled=%s", enabled)
-            return jsonify({"enabled": enabled})
+            enabled = self._settings_service.update_chat_whitelist_enabled(payload.get('enabled'))
+            self._logger.info('Chat whitelist updated: enabled=%s', enabled)
+            return jsonify({'enabled': enabled})
         except ValueError as exc:
-            return jsonify({"error": str(exc)}), 400
+            return jsonify({'error': str(exc)}), 400
         except Exception as exc:
-            self._logger.error(f"Error updating chat whitelist: {exc}")
-            return jsonify({"error": str(exc)}), 500
+            self._logger.error('Error updating chat whitelist: %s', exc)
+            return jsonify({'error': str(exc)}), 500

src/services/__init__.py

@@ -4,8 +4,18 @@
 
 from .authentication_service import AuthenticationService
 from .log_service import LogService
+from .model_discovery_service import ModelDiscoveryService
 from .provider_service import ProviderService
 from .proxy_service import ProxyService
+from .settings_service import SettingsService
 from .user_service import UserService
 
-__all__ = ["AuthenticationService", "UserService", "LogService", "ProxyService", "ProviderService"]
+__all__ = [
+    'AuthenticationService',
+    'UserService',
+    'LogService',
+    'ModelDiscoveryService',
+    'ProxyService',
+    'ProviderService',
+    'SettingsService',
+]

src/services/model_discovery_service.py

@@ -0,0 +1,155 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""上游模型发现服务。"""
+
+from __future__ import annotations
+
+from typing import Any, Dict, List, Optional
+from urllib.parse import urlparse
+
+import requests
+
+from ..application.app_context import AppContext
+from ..config.provider_config import clean_optional_string, parse_optional_bool, parse_optional_positive_int
+from ..utils.net import build_requests_proxies, normalize_proxy_url
+
+
+class ModelDiscoveryService:
+    """负责探测上游 provider 的模型列表。"""
+
+    def __init__(self, ctx: AppContext):
+        self._logger = ctx.logger
+
+    def fetch_models_preview(
+        self,
+        api: str,
+        api_key: Optional[str] = None,
+        proxy: Optional[str] = None,
+        timeout_seconds: Optional[Any] = None,
+        verify_ssl: Optional[Any] = None,
+    ) -> Dict[str, Any]:
+        if not api or not str(api).strip():
+            raise ValueError('Provider api is required')
+
+        fetched_models = self._fetch_models_from_upstream(
+            api=str(api).strip(),
+            api_key=clean_optional_string(api_key),
+            proxy=normalize_proxy_url(proxy),
+            timeout_seconds=parse_optional_positive_int(timeout_seconds, default=30) or 30,
+            verify_ssl=parse_optional_bool(verify_ssl, default=False) or False,
+        )
+
+        return {
+            'fetched_models': fetched_models,
+            'fetched_count': len(fetched_models),
+        }
+
+    def _fetch_models_from_upstream(
+        self,
+        api: str,
+        api_key: Optional[str],
+        proxy: Optional[str],
+        timeout_seconds: int,
+        verify_ssl: bool,
+    ) -> List[str]:
+        headers = {'accept': 'application/json'}
+        if api_key:
+            headers['authorization'] = f'Bearer {api_key}'
+        proxies = build_requests_proxies(proxy)
+
+        candidates = self._build_model_endpoint_candidates(api)
+        last_error: Optional[str] = None
+
+        with requests.Session() as session:
+            for url in candidates:
+                try:
+                    response = session.get(
+                        url,
+                        headers=headers,
+                        proxies=proxies,
+                        timeout=timeout_seconds,
+                        verify=verify_ssl,
+                    )
+                    if response.status_code >= 400:
+                        last_error = f'{url} returned {response.status_code}'
+                        continue
+
+                    payload = response.json()
+                    models = self._extract_models_from_payload(payload)
+                    if models:
+                        self._logger.info(
+                            'Fetched %s models from provider endpoint: provider_api=%s endpoint=%s',
+                            len(models),
+                            api,
+                            url,
+                        )
+                        return models
+                    last_error = f'{url} returned no models'
+                except requests.RequestException as exc:
+                    last_error = f'{url} request failed: {exc}'
+                except ValueError as exc:
+                    last_error = f'{url} returned invalid json: {exc}'
+
+        raise ValueError(last_error or 'Failed to fetch models')
+
+    @staticmethod
+    def _build_model_endpoint_candidates(api: str) -> List[str]:
+        cleaned_api = api.strip().rstrip('/')
+        parsed = urlparse(cleaned_api)
+        if not parsed.scheme or not parsed.netloc:
+            raise ValueError('Provider api must be a valid absolute URL')
+
+        root = f'{parsed.scheme}://{parsed.netloc}'
+        path = parsed.path.rstrip('/')
+        path_prefixes = ['']
+
+        if path:
+            path_parts = [segment for segment in path.split('/') if segment]
+            for cut in range(len(path_parts), 0, -1):
+                current = '/' + '/'.join(path_parts[:cut])
+                lower_current = current.lower()
+                if lower_current.endswith('/chat/completions'):
+                    current = current[: -len('/chat/completions')]
+                elif lower_current.endswith('/completions'):
+                    current = current[: -len('/completions')]
+                path_prefixes.append(current.rstrip('/'))
+            path_prefixes.append(path)
+
+        candidates: List[str] = []
+        seen: set[str] = set()
+        for prefix in path_prefixes:
+            normalized_prefix = prefix.rstrip('/')
+            for suffix in ('/v1/models', '/models'):
+                candidate = f'{root}{normalized_prefix}{suffix}'
+                if candidate in seen:
+                    continue
+                seen.add(candidate)
+                candidates.append(candidate)
+        return candidates
+
+    @staticmethod
+    def _extract_models_from_payload(payload: Any) -> List[str]:
+        models: List[str] = []
+        items: Any = None
+        if isinstance(payload, dict):
+            items = payload.get('data')
+            if items is None and isinstance(payload.get('models'), list):
+                items = payload.get('models')
+        elif isinstance(payload, list):
+            items = payload
+
+        if not isinstance(items, list):
+            return models
+
+        seen: set[str] = set()
+        for item in items:
+            if isinstance(item, dict):
+                model = item.get('id') or item.get('name')
+            else:
+                model = item
+            model_name = str(model).strip() if model is not None else ''
+            if not model_name or model_name in seen:
+                continue
+            seen.add(model_name)
+            models.append(model_name)
+        return models

src/services/provider_service.py

@@ -4,434 +4,97 @@
 
 from __future__ import annotations
 
-import os
-import tempfile
-from pathlib import Path
-from typing import Any, Callable, Dict, List, Optional
-from urllib.parse import urlparse
-
-import requests
-import yaml
+from typing import Callable, Dict, List, Optional
 
 from ..application.app_context import AppContext
-from ..config import ProviderManager
-from ..utils.net import build_requests_proxies, normalize_proxy_url
+from ..config.provider_config import (
+    normalize_provider_payload,
+    validate_provider_definitions,
+)
 
 
 class ProviderService:
-    """负责 provider 配置的增删改查、写回与模型拉取。"""
+    """负责 provider 配置的增删改查与模型拉取。"""
 
-    def __init__(self, ctx: AppContext, config_path: Path, reload_callback: Callable[[], None]):
-        self._ctx = ctx
+    def __init__(self, ctx: AppContext, reload_callback: Callable[[], None]):
+        self._config_manager = ctx.config_manager
         self._logger = ctx.logger
-        self._config_path = Path(config_path).resolve()
         self._reload_callback = reload_callback
 
     def list_providers(self) -> List[Dict[str, Any]]:
-        config = self._load_config()
-        providers = self._extract_providers(config)
-        return [self._clone_provider(provider) for provider in providers]
+        config = self._config_manager.get_raw_config()
+        return self._extract_providers(config)
 
     def get_provider(self, name: str) -> Optional[Dict[str, Any]]:
-        provider = self._find_provider(self.list_providers(), name)
-        return self._clone_provider(provider) if provider else None
+        config = self._config_manager.get_raw_config()
+        providers = self._extract_providers(config)
+        return self._find_provider(providers, name)
 
     def create_provider(self, payload: Dict[str, Any]) -> Dict[str, Any]:
-        config = self._load_config()
+        config = self._config_manager.get_raw_config()
         providers = self._extract_providers(config)
-        normalized = self._normalize_provider_payload(payload)
+        normalized = normalize_provider_payload(payload)
 
-        if self._find_provider(providers, normalized["name"]):
+        if self._find_provider(providers, normalized['name']):
             raise ValueError(f"Provider already exists: {normalized['name']}")
 
         providers.append(normalized)
-        self._validate_providers(providers)
-        config["providers"] = providers
-        self._write_config(config)
-        self._reload_callback()
-        return self._clone_provider(normalized)
+        self._save_providers(config, providers)
+        return normalized
 
     def update_provider(self, current_name: str, payload: Dict[str, Any]) -> Dict[str, Any]:
-        config = self._load_config()
+        config = self._config_manager.get_raw_config()
         providers = self._extract_providers(config)
-        normalized = self._normalize_provider_payload(payload)
+        normalized = normalize_provider_payload(payload)
 
         target = self._find_provider(providers, current_name)
         if target is None:
-            raise ValueError(f"Provider not found: {current_name}")
+            raise ValueError(f'Provider not found: {current_name}')
 
-        duplicate = self._find_provider(providers, normalized["name"])
+        duplicate = self._find_provider(providers, normalized['name'])
         if duplicate is not None and duplicate is not target:
             raise ValueError(f"Provider already exists: {normalized['name']}")
 
-        target_index = providers.index(target)
-        providers[target_index] = normalized
-        self._validate_providers(providers)
-        config["providers"] = providers
-        self._write_config(config)
-        self._reload_callback()
-        return self._clone_provider(normalized)
+        providers[providers.index(target)] = normalized
+        self._save_providers(config, providers)
+        return normalized
 
     def delete_provider(self, name: str) -> None:
-        config = self._load_config()
+        config = self._config_manager.get_raw_config()
         providers = self._extract_providers(config)
         target = self._find_provider(providers, name)
         if target is None:
-            raise ValueError(f"Provider not found: {name}")
+            raise ValueError(f'Provider not found: {name}')
 
         providers.remove(target)
+        self._save_providers(config, providers)
+
+    def _save_providers(self, config: Dict[str, Any], providers: List[Dict[str, Any]]) -> None:
         self._validate_providers(providers)
-        config["providers"] = providers
-        self._write_config(config)
+        config['providers'] = providers
+        self._config_manager.write_raw_config(config)
         self._reload_callback()
 
-    def fetch_models_preview(
-        self,
-        api: str,
-        api_key: Optional[str] = None,
-        proxy: Optional[str] = None,
-        timeout_seconds: Optional[Any] = None,
-        verify_ssl: Optional[Any] = None,
-    ) -> Dict[str, Any]:
-        if not api or not str(api).strip():
-            raise ValueError("Provider api is required")
-
-        effective_api_key = self._clean_optional_string(api_key)
-        effective_proxy = self._parse_optional_proxy(proxy)
-
-        effective_timeout = self._parse_optional_positive_int(timeout_seconds)
-        if effective_timeout is None:
-            effective_timeout = 30
-
-        effective_verify_ssl = self._parse_optional_bool(verify_ssl)
-        if effective_verify_ssl is None:
-            effective_verify_ssl = False
-
-        fetched_models = self._fetch_models_from_upstream(
-            api=str(api).strip(),
-            api_key=effective_api_key,
-            proxy=effective_proxy,
-            timeout_seconds=effective_timeout,
-            verify_ssl=effective_verify_ssl,
-        )
-
-        return {
-            "fetched_models": fetched_models,
-            "fetched_count": len(fetched_models),
-        }
-
-    def update_chat_whitelist_enabled(self, enabled: Any) -> bool:
-        parsed_enabled = self._parse_optional_bool(enabled)
-        if parsed_enabled is None:
-            raise ValueError("Whitelist enabled flag is required")
-
-        config = self._load_config()
-        chat_config = config.get("chat")
-        if chat_config is None:
-            chat_config = {}
-            config["chat"] = chat_config
-        if not isinstance(chat_config, dict):
-            raise ValueError("Config field 'chat' must be an object")
-
-        chat_config["whitelist_enabled"] = parsed_enabled
-        self._write_config(config)
-        self._reload_callback()
-        return parsed_enabled
-
-    def _load_config(self) -> Dict[str, Any]:
-        with open(self._config_path, "r", encoding="utf-8") as file:
-            data = yaml.safe_load(file)
-
-        if data is None:
-            return {}
-        if not isinstance(data, dict):
-            raise ValueError("Configuration file must contain a top-level mapping")
-        return data
-
-    def _write_config(self, config: Dict[str, Any]) -> None:
-        temp_file_path: Optional[str] = None
-        try:
-            with tempfile.NamedTemporaryFile(
-                "w",
-                encoding="utf-8",
-                dir=self._config_path.parent,
-                delete=False,
-            ) as temp_file:
-                yaml.safe_dump(config, temp_file, allow_unicode=True, sort_keys=False)
-                temp_file.flush()
-                os.fsync(temp_file.fileno())
-                temp_file_path = temp_file.name
-
-            os.replace(temp_file_path, self._config_path)
-        finally:
-            if temp_file_path and os.path.exists(temp_file_path):
-                os.remove(temp_file_path)
-
     @staticmethod
     def _extract_providers(config: Dict[str, Any]) -> List[Dict[str, Any]]:
-        providers = config.get("providers", [])
+        providers = config.get('providers', [])
         if providers is None:
             providers = []
         if not isinstance(providers, list):
             raise ValueError("Config field 'providers' must be a list")
         for index, provider in enumerate(providers):
             if not isinstance(provider, dict):
-                raise ValueError(f"Provider entry at index {index} must be an object")
+                raise ValueError(f'Provider entry at index {index} must be an object')
         return list(providers)
 
     @staticmethod
     def _find_provider(providers: List[Dict[str, Any]], name: str) -> Optional[Dict[str, Any]]:
         normalized_name = str(name).strip()
         for provider in providers:
-            if str(provider.get("name", "")).strip() == normalized_name:
+            if str(provider.get('name', '')).strip() == normalized_name:
                 return provider
         return None
 
-    def _validate_providers(self, providers: List[Dict[str, Any]]) -> None:
-        manager = ProviderManager(self._ctx)
-        manager.load_providers([self._clone_provider(provider) for provider in providers])
-
-    def _normalize_provider_payload(self, payload: Dict[str, Any]) -> Dict[str, Any]:
-        name = str(payload.get("name", "")).strip()
-        api = str(payload.get("api", "")).strip()
-        if not name:
-            raise ValueError("Provider name is required")
-        if not api:
-            raise ValueError("Provider api is required")
-
-        normalized: Dict[str, Any] = {
-            "name": name,
-            "api": api,
-        }
-
-        api_key = self._clean_optional_string(payload.get("api_key"))
-        if api_key is not None:
-            normalized["api_key"] = api_key
-
-        proxy = self._parse_optional_proxy(payload.get("proxy"))
-        if proxy is not None:
-            normalized["proxy"] = proxy
-
-        timeout_seconds = self._parse_optional_positive_int(payload.get("timeout_seconds"))
-        if timeout_seconds is not None:
-            normalized["timeout_seconds"] = timeout_seconds
-
-        max_retries = self._parse_optional_positive_int(payload.get("max_retries"))
-        if max_retries is not None:
-            normalized["max_retries"] = max_retries
-
-        verify_ssl = self._parse_optional_bool(payload.get("verify_ssl"))
-        if verify_ssl is not None:
-            normalized["verify_ssl"] = verify_ssl
-
-        model_list = self._normalize_model_list(payload.get("model_list"))
-        if model_list:
-            normalized["model_list"] = model_list
-
-        hook = self._clean_optional_string(payload.get("hook"))
-        if hook is not None:
-            normalized["hook"] = hook
-
-        return normalized
-
     @staticmethod
-    def _clean_optional_string(value: Any) -> Optional[str]:
-        if value is None:
-            return None
-        text = str(value).strip()
-        return text or None
-
-    @staticmethod
-    def _parse_optional_positive_int(value: Any) -> Optional[int]:
-        if value is None:
-            return None
-        if isinstance(value, str) and not value.strip():
-            return None
-        try:
-            parsed = int(value)
-        except (TypeError, ValueError) as exc:
-            raise ValueError("Expected a positive integer") from exc
-        if parsed <= 0:
-            raise ValueError("Expected a positive integer")
-        return parsed
-
-    @staticmethod
-    def _parse_optional_proxy(value: Any) -> Optional[str]:
-        if value is None:
-            return None
-        return normalize_proxy_url(value)
-
-    @staticmethod
-    def _parse_optional_bool(value: Any) -> Optional[bool]:
-        if value is None:
-            return None
-        if isinstance(value, str):
-            lowered = value.strip().lower()
-            if lowered == "":
-                return None
-            if lowered in {"true", "1", "yes", "on"}:
-                return True
-            if lowered in {"false", "0", "no", "off"}:
-                return False
-            raise ValueError("Expected a boolean value")
-        if isinstance(value, bool):
-            return value
-        if isinstance(value, (int, float)):
-            return bool(value)
-        raise ValueError("Expected a boolean value")
-
-    def _normalize_model_list(self, value: Any) -> List[str]:
-        models: List[str] = []
-        if value is None:
-            return models
-
-        if isinstance(value, str):
-            raw_items = value.replace(",", "\n").splitlines()
-        elif isinstance(value, list):
-            raw_items = value
-        else:
-            raise ValueError("model_list must be a list or newline-separated string")
-
-        seen: set[str] = set()
-        for item in raw_items:
-            model = str(item).strip()
-            if not model:
-                continue
-            if model in seen:
-                continue
-            seen.add(model)
-            models.append(model)
-        return models
-
-    @staticmethod
-    def _merge_models(existing: List[str], fetched: List[str]) -> List[str]:
-        merged = list(existing)
-        seen = set(existing)
-        for model in fetched:
-            if model in seen:
-                continue
-            seen.add(model)
-            merged.append(model)
-        return merged
-
-    def _fetch_models_from_upstream(
-        self,
-        api: str,
-        api_key: Optional[str],
-        proxy: Optional[str],
-        timeout_seconds: int,
-        verify_ssl: bool,
-    ) -> List[str]:
-        headers = {"accept": "application/json"}
-        if api_key:
-            headers["authorization"] = f"Bearer {api_key}"
-        proxies = build_requests_proxies(proxy)
-
-        candidates = self._build_model_endpoint_candidates(api)
-        last_error: Optional[str] = None
-
-        with requests.Session() as session:
-            for url in candidates:
-                try:
-                    response = session.get(
-                        url,
-                        headers=headers,
-                        proxies=proxies,
-                        timeout=timeout_seconds,
-                        verify=verify_ssl,
-                    )
-                    if response.status_code >= 400:
-                        last_error = f"{url} returned {response.status_code}"
-                        continue
-
-                    payload = response.json()
-                    models = self._extract_models_from_payload(payload)
-                    if models:
-                        self._logger.info(
-                            "Fetched %s models from provider endpoint: provider_api=%s endpoint=%s",
-                            len(models),
-                            api,
-                            url,
-                        )
-                        return models
-                    last_error = f"{url} returned no models"
-                except requests.RequestException as exc:
-                    last_error = f"{url} request failed: {exc}"
-                except ValueError as exc:
-                    last_error = f"{url} returned invalid json: {exc}"
-
-        raise ValueError(last_error or "Failed to fetch models")
-
-    @staticmethod
-    def _build_model_endpoint_candidates(api: str) -> List[str]:
-        cleaned_api = api.strip().rstrip("/")
-        parsed = urlparse(cleaned_api)
-        if not parsed.scheme or not parsed.netloc:
-            raise ValueError("Provider api must be a valid absolute URL")
-
-        root = f"{parsed.scheme}://{parsed.netloc}"
-        path = parsed.path.rstrip("/")
-        path_prefixes = [""]
-
-        if path:
-            path_parts = [segment for segment in path.split("/") if segment]
-            for cut in range(len(path_parts), 0, -1):
-                current = "/" + "/".join(path_parts[:cut])
-                lower_current = current.lower()
-                if lower_current.endswith("/chat/completions"):
-                    current = current[: -len("/chat/completions")]
-                elif lower_current.endswith("/completions"):
-                    current = current[: -len("/completions")]
-                path_prefixes.append(current.rstrip("/"))
-            path_prefixes.append(path)
-
-        candidates: List[str] = []
-        seen: set[str] = set()
-        for prefix in path_prefixes:
-            normalized_prefix = prefix.rstrip("/")
-            for suffix in ("/v1/models", "/models"):
-                candidate = f"{root}{normalized_prefix}{suffix}"
-                if candidate in seen:
-                    continue
-                seen.add(candidate)
-                candidates.append(candidate)
-        return candidates
-
-    @staticmethod
-    def _extract_models_from_payload(payload: Any) -> List[str]:
-        models: List[str] = []
-        items: Any = None
-        if isinstance(payload, dict):
-            items = payload.get("data")
-            if items is None and isinstance(payload.get("models"), list):
-                items = payload.get("models")
-        elif isinstance(payload, list):
-            items = payload
-
-        if not isinstance(items, list):
-            return models
-
-        seen: set[str] = set()
-        for item in items:
-            if isinstance(item, dict):
-                model = item.get("id") or item.get("name")
-            else:
-                model = item
-            model_name = str(model).strip() if model is not None else ""
-            if not model_name or model_name in seen:
-                continue
-            seen.add(model_name)
-            models.append(model_name)
-        return models
-
-    @staticmethod
-    def _clone_provider(provider: Dict[str, Any]) -> Dict[str, Any]:
-        cloned: Dict[str, Any] = {}
-        for key, value in provider.items():
-            if isinstance(value, list):
-                cloned[key] = list(value)
-            elif isinstance(value, dict):
-                cloned[key] = dict(value)
-            else:
-                cloned[key] = value
-        return cloned
+    def _validate_providers(providers: List[Dict[str, Any]]) -> None:
+        validate_provider_definitions(providers)

src/services/settings_service.py

@@ -0,0 +1,32 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""通用配置项维护服务。"""
+
+from typing import Any
+
+from ..application.app_context import AppContext
+from ..config.provider_config import parse_optional_bool
+
+
+class SettingsService:
+    """负责非 provider 专属的配置项更新。"""
+
+    def __init__(self, ctx: AppContext):
+        self._config_manager = ctx.config_manager
+
+    def update_chat_whitelist_enabled(self, enabled: Any) -> bool:
+        parsed_enabled = parse_optional_bool(enabled)
+        if parsed_enabled is None:
+            raise ValueError("Whitelist enabled flag is required")
+
+        config = self._config_manager.get_raw_config()
+        chat_config = config.get("chat")
+        if chat_config is None:
+            chat_config = {}
+            config["chat"] = chat_config
+        if not isinstance(chat_config, dict):
+            raise ValueError("Config field 'chat' must be an object")
+
+        chat_config["whitelist_enabled"] = parsed_enabled
+        self._config_manager.write_raw_config(config)
+        return parsed_enabled