refactor: expand chain runtime config snapshot

app/application/configuration.py

@@ -4,7 +4,7 @@ from __future__ import annotations
 
 from collections.abc import Callable
 from dataclasses import dataclass
-from typing import Any, Protocol
+from typing import Any, Optional, Protocol
 
 
 class SystemConfigReader(Protocol):
@@ -80,6 +80,12 @@ class ChainRuntimeConfig:
     """Chain 在一次宿主生命周期内使用的基础配置快照。"""
 
     media_extensions: tuple[str, ...]
+    superuser: str = "admin"
+    media_recognize_share: bool = False
+    auxiliary_auth_enable: bool = False
+    global_image_cache: bool = False
+    auto_download_user: Optional[str] = None
+    resource_url: Optional[str] = None
 
 
 @dataclass(frozen=True, slots=True)

app/chain/_messaging.py

@@ -13,7 +13,6 @@ from app.domain.context import Context, MediaInfo, MusicInfo, TorrentInfo
 from app.domain.meta.metabase import MetaBase
 from app.foundation.identity import normalize_internal_user_id
 from app.application.messaging.message import MessageTemplateHelper
-from app.runtime.config import settings
 from app.runtime.extensions.service_config import ServiceConfigHelper
 from app.runtime.log import logger
 from app.schemas.message import MessageResponse
@@ -176,7 +175,9 @@ class NotificationMixin:
                         # 仅发送管理员
                         logger.info(f"{send_message.mtype} 的消息已设置发送给管理员")
                         # 读取管理员消息IDS
-                        send_message.targets = useroper.get_settings(settings.SUPERUSER)
+                        send_message.targets = useroper.get_settings(
+                            self.runtime_config.superuser
+                        )
                         admin_sended = True
                     elif action == "user" and send_message.username:
                         # 发送对应用户
@@ -196,7 +197,7 @@ class NotificationMixin:
                                 )
                                 # 读取管理员消息IDS
                                 send_message.targets = useroper.get_settings(
-                                    settings.SUPERUSER
+                                    self.runtime_config.superuser
                                 )
                                 admin_sended = True
                             else:
@@ -205,7 +206,7 @@ class NotificationMixin:
                                     f"用户 {send_message.username} 不存在，消息无法发送到对应用户"
                                 )
                                 continue
-                        elif send_message.username == settings.SUPERUSER:
+                        elif send_message.username == self.runtime_config.superuser:
                             # 管理员同名已发送
                             admin_sended = True
                     else:
@@ -292,7 +293,9 @@ class NotificationMixin:
                         # 仅发送管理员
                         logger.info(f"{send_message.mtype} 的消息已设置发送给管理员")
                         # 读取管理员消息IDS
-                        send_message.targets = useroper.get_settings(settings.SUPERUSER)
+                        send_message.targets = useroper.get_settings(
+                            self.runtime_config.superuser
+                        )
                         admin_sended = True
                     elif action == "user" and send_message.username:
                         # 发送对应用户
@@ -312,7 +315,7 @@ class NotificationMixin:
                                 )
                                 # 读取管理员消息IDS
                                 send_message.targets = useroper.get_settings(
-                                    settings.SUPERUSER
+                                    self.runtime_config.superuser
                                 )
                                 admin_sended = True
                             else:
@@ -321,7 +324,7 @@ class NotificationMixin:
                                     f"用户 {send_message.username} 不存在，消息无法发送到对应用户"
                                 )
                                 continue
-                        elif send_message.username == settings.SUPERUSER:
+                        elif send_message.username == self.runtime_config.superuser:
                             # 管理员同名已发送
                             admin_sended = True
                     else:

app/chain/_recognition.py

@@ -15,7 +15,6 @@ from app.domain.context import MediaInfo, MusicInfo
 from app.domain.meta.metabase import MetaBase
 from app.domain.meta.metamusic import MetaMusic
 from app.runtime.cache import fresh, async_fresh
-from app.runtime.config import settings
 from app.runtime.events import Event
 from app.runtime.log import logger
 from app.schemas.media import normalize_media_source, resolve_media_identity
@@ -24,8 +23,8 @@ from app.schemas.types import ChainEventType, MediaSource, MediaType, SystemConf
 
 class RecognitionMixin:
 
-    @staticmethod
     def _can_use_media_recognize_share(
+            self,
             meta: Optional[MetaBase],
             media_source: Optional[MediaSource],
             media_id: Optional[str],
@@ -34,7 +33,7 @@ class RecognitionMixin:
         仅在名称识别场景下使用共享识别，显式ID识别不再重复回查
         """
         return bool(
-            settings.MEDIA_RECOGNIZE_SHARE
+            self.runtime_config.media_recognize_share
             and meta
             and not media_source
             and not media_id
@@ -515,4 +514,3 @@ class RecognitionMixin:
             f"（{plugin_info.media_source}:{plugin_info.media_id}）"
         )
         return plugin_info
-

app/chain/interaction.py

@@ -19,7 +19,6 @@ from app.domain import title as title_rules
 from app.domain.context import Context, MediaInfo
 from app.domain.meta.metabase import MetaBase
 from app.foundation import url as url_tools
-from app.runtime.config import settings
 from app.runtime.log import logger
 from app.schemas.download import DownloadDirectory
 from app.schemas.file import FileURI
@@ -1130,7 +1129,7 @@ class MediaInteractionChain(ChainBase):
                 source=source,
                 title=title,
                 userid=userid,
-                link=settings.MP_DOMAIN("#/resource"),
+                link=self.runtime_config.resource_url,
                 buttons=buttons,
                 original_message_id=original_message_id,
                 original_chat_id=original_chat_id,
@@ -1533,12 +1532,11 @@ class MediaInteractionChain(ChainBase):
             for sea, no_exist in season_map.items()
         ]
 
-    @staticmethod
-    def _should_auto_download(userid: Union[str, int]) -> bool:
+    def _should_auto_download(self, userid: Union[str, int]) -> bool:
         """
         判断当前用户是否命中自动下载名单。
         """
-        auto_download_user = settings.AUTO_DOWNLOAD_USER
+        auto_download_user = self.runtime_config.auto_download_user
         return bool(
             auto_download_user
             and (

app/chain/recommend.py

@@ -8,7 +8,7 @@ from app.chain.douban import DoubanChain
 from app.chain.listenbrainz import ListenBrainzChain
 from app.chain.tmdb import TmdbChain
 from app.runtime.cache import cached, fresh
-from app.runtime.config import settings, global_vars
+from app.runtime.config import global_vars
 from app.domain.context import MusicInfo
 from app.application.image import ImageHelper
 from app.runtime.log import logger
@@ -272,7 +272,7 @@ class RecommendChain(ChainBase, metaclass=Singleton):
         :param datas: 数据列表
         :param progress_callback: 定时服务进度更新回调
         """
-        if not settings.GLOBAL_IMAGE_CACHE:
+        if not self.runtime_config.global_image_cache:
             return
 
         total_num = len(datas)

app/chain/user.py

@@ -3,7 +3,6 @@ from dataclasses import dataclass
 from typing import Any, Literal, Optional, Tuple, Union
 
 from app.chain import ChainBase
-from app.runtime.config import settings
 from app.application.security.token import get_password_hash, verify_password
 from app.application.chain.data import UserPortProxy as UserOper
 from app.runtime.log import logger
@@ -73,7 +72,7 @@ class UserChain(ChainBase):
                 return True, user_or_message
             else:
                 # 用户不存在或密码错误，考虑辅助认证
-                if settings.AUXILIARY_AUTH_ENABLE:
+                if self.runtime_config.auxiliary_auth_enable:
                     logger.warning("密码认证失败，尝试通过外部服务进行辅助认证 ...")
                     aux_success, aux_user_or_message = self.auxiliary_authenticate(credentials=credentials)
                     if aux_success:
@@ -91,7 +90,7 @@ class UserChain(ChainBase):
                     return False, PASSWORD_INVALID_CREDENTIALS_MESSAGE
         elif credentials.grant_type == "authorization_code":
             # 处理其他认证类型的分支
-            if settings.AUXILIARY_AUTH_ENABLE:
+            if self.runtime_config.auxiliary_auth_enable:
                 aux_success, aux_user_or_message = self.auxiliary_authenticate(credentials=credentials)
                 if aux_success:
                     return True, aux_user_or_message

app/startup/modules_initializer.py

@@ -207,14 +207,20 @@ def _build_scheduler_runtime_config() -> SchedulerRuntimeConfig:
 
 
 def _build_chain_runtime_config() -> ChainRuntimeConfig:
-    """构建 Chain 通用媒体文件后缀配置快照。"""
+    """构建 Chain 在本次实例生命周期内使用的部署配置快照。"""
     return ChainRuntimeConfig(
         media_extensions=tuple(
             settings.RMT_MEDIAEXT
             + settings.DOWNLOAD_TMPEXT
             + settings.RMT_SUBEXT
             + settings.RMT_AUDIOEXT
-        )
+        ),
+        superuser=settings.SUPERUSER,
+        media_recognize_share=settings.MEDIA_RECOGNIZE_SHARE,
+        auxiliary_auth_enable=settings.AUXILIARY_AUTH_ENABLE,
+        global_image_cache=settings.GLOBAL_IMAGE_CACHE,
+        auto_download_user=settings.AUTO_DOWNLOAD_USER,
+        resource_url=settings.MP_DOMAIN("#/resource"),
     )