From c1f50f6d3f9969f4e1b588bb47d26e30a74715a3 Mon Sep 17 00:00:00 2001 From: jxxghp Date: Sat, 29 Aug 2026 19:58:33 +0800 Subject: [PATCH] refactor(runtime): unify paired decision flows --- app/chain/torrents.py | 69 ++++++++++---------- app/runtime/event/dispatch.py | 71 ++++++++++++--------- app/runtime/extensions/plugin/dependency.py | 48 +++++++------- 3 files changed, 101 insertions(+), 87 deletions(-) diff --git a/app/chain/torrents.py b/app/chain/torrents.py index 027f9aaac..380682b29 100644 --- a/app/chain/torrents.py +++ b/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, diff --git a/app/runtime/event/dispatch.py b/app/runtime/event/dispatch.py index 75fee5551..26ae0d101 100644 --- a/app/runtime/event/dispatch.py +++ b/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: """按订阅快照把广播事件投递到线程池或主事件循环。""" diff --git a/app/runtime/extensions/plugin/dependency.py b/app/runtime/extensions/plugin/dependency.py index 31d3351bb..d5bdd418c 100644 --- a/app/runtime/extensions/plugin/dependency.py +++ b/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: """分类物理插件，并把源码结论映射到全部虚拟实例。"""