refactor(runtime): unify paired decision flows

app/chain/torrents.py

@@ -32,6 +32,27 @@ class TorrentsChain(ChainBase):
     _music_spider_file = "__torrents_music_cache__"
     _music_rss_file = "__rss_music_cache__"
 
+    def _cache_type(self, stype: Optional[str]) -> str:
+        """把可选缓存类型统一为本次读取使用的明确订阅模式。"""
+        return stype or self.runtime_config.subscribe_mode
+
+    def _cache_file_pair(self, stype: str) -> tuple[str, str]:
+        """返回指定订阅模式下影视与音乐缓存的唯一文件映射。"""
+        if stype == 'spider':
+            return self._spider_file, self._music_spider_file
+        return self._rss_file, self._music_rss_file
+
+    @staticmethod
+    def _merge_torrent_caches(
+        torrents_cache: Dict[str, List[Context]],
+        music_cache: Dict[str, List[Context]],
+    ) -> Dict[str, List[Context]]:
+        """按站点稳定合并音乐独立缓存，空列表不创建无效站点键。"""
+        for domain, contexts in music_cache.items():
+            if contexts:
+                torrents_cache.setdefault(domain, []).extend(contexts)
+        return torrents_cache
+
     @property
     def cache_file(self) -> str:
         """
@@ -63,34 +84,24 @@ class TorrentsChain(ChainBase):
         :param stype: 强制指定缓存类型，spider:爬虫缓存，rss:rss缓存
         """
 
-        if not stype:
-            stype = self.runtime_config.subscribe_mode
-
-        # 读取缓存
-        if stype == 'spider':
-            torrents_cache = self.load_cache(self._spider_file) or {}
-        else:
-            torrents_cache = self.load_cache(self._rss_file) or {}
+        stype = self._cache_type(stype)
+        video_file, _music_file = self._cache_file_pair(stype)
+        torrents_cache = self.load_cache(video_file) or {}
 
         # 兼容性处理：为旧版本的Context对象补齐新增候选识别字段
         self._ensure_context_compatibility(torrents_cache, stype=stype)
 
         # 合并音乐独立缓存，供订阅匹配等消费方按站点读取完整候选
         music_cache = self.get_music_torrents(stype=stype)
-        for domain, contexts in music_cache.items():
-            if contexts:
-                torrents_cache.setdefault(domain, []).extend(contexts)
-
-        return torrents_cache
+        return self._merge_torrent_caches(torrents_cache, music_cache)
 
     def get_music_torrents(self, stype: Optional[str] = None) -> Dict[str, List[Context]]:
         """
         获取音乐独立缓存的种子
         :param stype: 强制指定缓存类型，spider:爬虫缓存，rss:rss缓存
         """
-        if not stype:
-            stype = self.runtime_config.subscribe_mode
-        music_file = self._music_spider_file if stype == 'spider' else self._music_rss_file
+        stype = self._cache_type(stype)
+        _video_file, music_file = self._cache_file_pair(stype)
         music_cache = self.load_cache(music_file) or {}
         # 兼容性处理：为旧版本的Context对象补齐新增候选识别字段
         self._ensure_context_compatibility(music_cache, stype=stype)
@@ -101,11 +112,7 @@ class TorrentsChain(ChainBase):
         返回影视与音乐缓存文件名，供按当前订阅模式回写各自缓存
         :param stype: 强制指定缓存类型，spider:爬虫缓存，rss:rss缓存
         """
-        if not stype:
-            stype = self.runtime_config.subscribe_mode
-        if stype == 'spider':
-            return self._spider_file, self._music_spider_file
-        return self._rss_file, self._music_rss_file
+        return self._cache_file_pair(self._cache_type(stype))
 
     @staticmethod
     def split_cache_contexts(
@@ -131,27 +138,17 @@ class TorrentsChain(ChainBase):
         :param stype: 强制指定缓存类型，spider:爬虫缓存，rss:rss缓存
         """
 
-        if not stype:
-            stype = self.runtime_config.subscribe_mode
-
-        # 异步读取缓存
-        if stype == 'spider':
-            torrents_cache = await self.async_load_cache(self._spider_file) or {}
-            music_cache = await self.async_load_cache(self._music_spider_file) or {}
-        else:
-            torrents_cache = await self.async_load_cache(self._rss_file) or {}
-            music_cache = await self.async_load_cache(self._music_rss_file) or {}
+        stype = self._cache_type(stype)
+        video_file, music_file = self._cache_file_pair(stype)
+        torrents_cache = await self.async_load_cache(video_file) or {}
+        music_cache = await self.async_load_cache(music_file) or {}
 
         # 兼容性处理：为旧版本的Context对象补齐新增候选识别字段
         self._ensure_context_compatibility(torrents_cache, stype=stype)
         self._ensure_context_compatibility(music_cache, stype=stype)
 
         # 合并音乐独立缓存，供订阅匹配等消费方按站点读取完整候选
-        for domain, contexts in music_cache.items():
-            if contexts:
-                torrents_cache.setdefault(domain, []).extend(contexts)
-
-        return torrents_cache
+        return self._merge_torrent_caches(torrents_cache, music_cache)
 
     def get_subscribe_cache_candidates(
             self,

app/runtime/event/dispatch.py

@@ -40,8 +40,10 @@ class EventDispatcher:
         self._async_handle_sink = async_handle_sink
         self._sync_handle_sink = sync_handle_sink
 
-    def dispatch_chain(self, event: Any) -> bool:
-        """同步按优先级顺序执行链式事件快照。"""
+    def _begin_chain_dispatch(
+        self, event: Any
+    ) -> tuple[tuple[str, int, Callable[..., Any]], ...]:
+        """冻结启用处理器快照，并统一空快照与开始生命周期记录。"""
         handlers = self._registry.chain_snapshot(event.event_type)
         enabled = tuple(
             (handler_id, priority, handler)
@@ -50,45 +52,56 @@ class EventDispatcher:
         )
         if not enabled:
             logger.debug("No enabled handlers found for chain event: %s", event)
-            return False
+            return ()
         self._log_lifecycle(event, "Started")
-        for _handler_id, priority, handler in enabled:
-            started_at = time.time()
-            self.invoke_sync(handler, event)
-            logger.debug(
-                "%s (Priority: %s), completed in %.3fs for event: %s",
-                EventRegistry.handler_identifier(handler),
-                priority,
-                time.time() - started_at,
-                event,
-            )
+        return enabled
+
+    @staticmethod
+    def _log_chain_handler_completed(
+        handler: Callable[..., Any],
+        priority: int,
+        started_at: float,
+        event: Any,
+    ) -> None:
+        """统一记录同步与异步处理器完成耗时和优先级。"""
+        logger.debug(
+            "%s (Priority: %s), completed in %.3fs for event: %s",
+            EventRegistry.handler_identifier(handler),
+            priority,
+            time.time() - started_at,
+            event,
+        )
+
+    def _complete_chain_dispatch(self, event: Any) -> bool:
+        """统一提交链式事件完成生命周期并返回已处理标识。"""
         self._log_lifecycle(event, "Completed")
         return True
 
+    def dispatch_chain(self, event: Any) -> bool:
+        """同步按优先级顺序执行链式事件快照。"""
+        enabled = self._begin_chain_dispatch(event)
+        if not enabled:
+            return False
+        for _handler_id, priority, handler in enabled:
+            started_at = time.time()
+            self.invoke_sync(handler, event)
+            EventDispatcher._log_chain_handler_completed(
+                handler, priority, started_at, event
+            )
+        return self._complete_chain_dispatch(event)
+
     async def async_dispatch_chain(self, event: Any) -> bool:
         """异步按优先级顺序执行链式事件快照。"""
-        handlers = self._registry.chain_snapshot(event.event_type)
-        enabled = tuple(
-            (handler_id, priority, handler)
-            for handler_id, (priority, handler) in handlers
-            if self._registry.is_handler_enabled(handler)
-        )
+        enabled = self._begin_chain_dispatch(event)
         if not enabled:
-            logger.debug("No enabled handlers found for chain event: %s", event)
             return False
-        self._log_lifecycle(event, "Started")
         for _handler_id, priority, handler in enabled:
             started_at = time.time()
             await self.invoke_async(handler, event)
-            logger.debug(
-                "%s (Priority: %s), completed in %.3fs for event: %s",
-                EventRegistry.handler_identifier(handler),
-                priority,
-                time.time() - started_at,
-                event,
+            EventDispatcher._log_chain_handler_completed(
+                handler, priority, started_at, event
             )
-        self._log_lifecycle(event, "Completed")
-        return True
+        return self._complete_chain_dispatch(event)
 
     def dispatch_broadcast(self, event: Any) -> None:
         """按订阅快照把广播事件投递到线程池或主事件循环。"""

app/runtime/extensions/plugin/dependency.py

@@ -44,17 +44,22 @@ class PluginDependencyService:
         self._registry = registry
         self._logger = log
 
-    def install_missing_with_status(self) -> PluginDependencyInstallResult:
-        """安装缺失依赖并返回安装器的明确结果。"""
-        installer = self._system().dependency
-        missing = installer.find_missing()
+    def _begin_missing_install(self, missing: list[str]) -> Optional[float]:
+        """统一无缺失短路、安装清单日志和耗时起点。"""
         if not missing:
-            return PluginDependencyInstallResult(missing=[], success=True)
+            return None
         self._logger.debug(f"检测到缺失的依赖项: {missing}")
         self._logger.info(f"开始安装缺失的依赖项，共 {len(missing)} 个...")
-        started = time.time()
-        success, _message = installer.install(missing)
-        elapsed = time.time() - started
+        return time.time()
+
+    def _complete_missing_install(
+        self,
+        missing: list[str],
+        success: bool,
+        started_at: float,
+    ) -> PluginDependencyInstallResult:
+        """统一安装结果、耗时和成功或失败日志分类。"""
+        elapsed = time.time() - started_at
         if success:
             self._logger.info(
                 f"已完成 {len(missing)} 个依赖项安装，总耗时：{elapsed:.2f} 秒"
@@ -65,6 +70,16 @@ class PluginDependencyService:
             )
         return PluginDependencyInstallResult(missing=missing, success=success)
 
+    def install_missing_with_status(self) -> PluginDependencyInstallResult:
+        """安装缺失依赖并返回安装器的明确结果。"""
+        installer = self._system().dependency
+        missing = installer.find_missing()
+        started_at = self._begin_missing_install(missing)
+        if started_at is None:
+            return PluginDependencyInstallResult(missing=[], success=True)
+        success, _message = installer.install(missing)
+        return self._complete_missing_install(missing, success, started_at)
+
     def install_missing(self) -> list[str]:
         """安装当前环境缺失的插件依赖并保持历史列表返回合同。"""
         return self.install_missing_with_status().missing
@@ -73,22 +88,11 @@ class PluginDependencyService:
         """在异步启动链中恢复缺失依赖，确保安装子进程可取消。"""
         installer = self._system().dependency
         missing = await installer.async_find_missing()
-        if not missing:
+        started_at = self._begin_missing_install(missing)
+        if started_at is None:
             return PluginDependencyInstallResult(missing=[], success=True)
-        self._logger.debug(f"检测到缺失的依赖项: {missing}")
-        self._logger.info(f"开始安装缺失的依赖项，共 {len(missing)} 个...")
-        started = time.time()
         success, _message = await installer.async_install(missing)
-        elapsed = time.time() - started
-        if success:
-            self._logger.info(
-                f"已完成 {len(missing)} 个依赖项安装，总耗时：{elapsed:.2f} 秒"
-            )
-        else:
-            self._logger.warning(
-                f"存在缺失依赖项安装失败，请尝试手动安装，总耗时：{elapsed:.2f} 秒"
-            )
-        return PluginDependencyInstallResult(missing=missing, success=success)
+        return self._complete_missing_install(missing, success, started_at)
 
     def classify_plugins(self) -> PluginDependencyClassification:
         """分类物理插件，并把源码结论映射到全部虚拟实例。"""