refactor: keep runtime settings below adapters

This commit is contained in:
jxxghp
2026-08-23 01:45:07 +08:00
parent d15c65b108
commit 2c6e8181c2
8 changed files with 47 additions and 39 deletions
+2 -7
View File
@@ -1,9 +1,8 @@
import json
import importlib
from typing import Any, Dict, Tuple, Optional
from app.application.configuration import get_runtime_settings
from app.runtime.log import logger
from app.runtime.settings import get_runtime_setting
from app.foundation.crypto import CryptoJsUtils, HashUtils
from app.adapters.network.http import RequestUtils
from app.domain import site as site_rules
@@ -13,11 +12,7 @@ from app.foundation.url import UrlUtils
def _runtime_setting(key: str) -> Any:
"""读取 CookieCloud 配置服务,未装配时回退旧 Settings ABI。"""
try:
return get_runtime_settings().get(key)
except RuntimeError:
legacy_settings = importlib.import_module("app.runtime.config").settings
return getattr(legacy_settings, key)
return get_runtime_setting(key)
class CookieCloudHelper:
+2 -7
View File
@@ -1,9 +1,8 @@
import base64
import importlib
from typing import Optional
from app.application.configuration import get_runtime_settings
from app.adapters.network.http import RequestUtils
from app.runtime.settings import get_runtime_setting
class OcrHelper:
@@ -14,11 +13,7 @@ class OcrHelper:
def __init__(self, ocr_base_url: Optional[str] = None) -> None:
"""初始化 OCR 服务地址,优先使用组合根设置快照。"""
if ocr_base_url is None:
try:
ocr_base_url = get_runtime_settings().get("OCR_HOST")
except RuntimeError:
legacy_settings = importlib.import_module("app.runtime.config").settings
ocr_base_url = legacy_settings.OCR_HOST
ocr_base_url = get_runtime_setting("OCR_HOST")
self._ocr_b64_url = f"{str(ocr_base_url).rstrip('/')}/captcha/base64"
def get_captcha_text(
+2 -7
View File
@@ -5,7 +5,6 @@ author: https://github.com/C5H12O5/syno-videoinfo-plugin
import base64
import concurrent
import concurrent.futures
import importlib
import json
import socket
import struct
@@ -14,9 +13,9 @@ import urllib.request
from threading import Lock
from typing import Dict, Optional
from app.application.configuration import get_runtime_settings
from app.runtime.log import logger
from app.runtime.reload import ConfigReloadMixin
from app.runtime.settings import get_runtime_setting
from app.foundation.singleton import Singleton
# DoH 关闭时需要释放线程池;保持惰性创建可避免未启用 DoH 时占用进程级资源
@@ -34,11 +33,7 @@ _orig_getaddrinfo = socket.getaddrinfo
def _doh_setting(key: str):
"""读取 DoH 热更新配置,组合根未装配时兼容旧 Settings。"""
try:
return get_runtime_settings().get(key)
except RuntimeError:
legacy_settings = importlib.import_module("app.runtime.config").settings
return getattr(legacy_settings, key)
return get_runtime_setting(key)
def _get_executor_locked() -> concurrent.futures.ThreadPoolExecutor:
+2 -7
View File
@@ -1,10 +1,9 @@
import importlib
import logging
from functools import lru_cache
from typing import List, Optional, Tuple
from app.application.configuration import get_runtime_settings
from app.runtime.log import logger, log_settings
from app.runtime.settings import get_runtime_setting
try:
import moviepilot_rust as _moviepilot_rust
@@ -17,11 +16,7 @@ else:
def _rust_accel_enabled() -> bool:
"""读取 Rust 开关快照,组合根未装配时回退旧 Settings。"""
try:
return bool(get_runtime_settings().get("RUST_ACCEL"))
except RuntimeError:
legacy_settings = importlib.import_module("app.runtime.config").settings
return bool(legacy_settings.RUST_ACCEL)
return bool(get_runtime_setting("RUST_ACCEL"))
def is_available() -> bool:
+25
View File
@@ -0,0 +1,25 @@
"""运行时配置读取端口,供低层适配器避免反向依赖 Application。"""
from __future__ import annotations
import importlib
from collections.abc import Callable
from typing import Any
RuntimeSettingProvider = Callable[[str], Any]
_provider: RuntimeSettingProvider | None = None
def configure_runtime_setting_provider(provider: RuntimeSettingProvider) -> None:
"""由启动组合根登记配置读取器,保持适配器只依赖 runtime 端口。"""
global _provider
_provider = provider
def get_runtime_setting(key: str) -> Any:
"""读取单项运行配置;启动早期未装配时回退旧 Settings ABI。"""
if _provider is not None:
return _provider(key)
legacy_settings = importlib.import_module("app.runtime.config").settings
return getattr(legacy_settings, key)
+2
View File
@@ -25,6 +25,7 @@ from app.runtime.extensions.plugin_manager import PluginManager
from app.runtime.events import EventHandlerBinding, EventManager
from app.runtime.observability import record_metric
from app.runtime.state import SystemHelper
from app.runtime.settings import configure_runtime_setting_provider
from app.runtime.thread import ThreadHelper
from app.adapters.network.doh import DohHelper
from app.adapters.system.resource import (
@@ -596,6 +597,7 @@ async def init_modules() -> HostRuntime:
)
configure_runtime_configuration(host_runtime.configuration)
configure_runtime_settings(host_runtime.settings)
configure_runtime_setting_provider(lambda key: getattr(settings, key))
configure_token_runtime_config(lambda: build_token_runtime_config(settings))
# 先发布系统配置服务,后续启动组合步骤统一复用同一配置端口。
configure_system_config(SystemConfigService(repository=SystemConfigOper()))