diff --git a/app/core/module.py b/app/core/module.py index 570ba637c..8f64e984e 100644 --- a/app/core/module.py +++ b/app/core/module.py @@ -58,12 +58,11 @@ class ModuleManager(metaclass=Singleton): """ logger.info("正在停止所有模块...") for module_id, module in self._running_modules.items(): - if hasattr(module, "stop"): - try: - module.stop() - logger.debug(f"Moudle Stoped:{module_id}") - except Exception as err: - logger.error(f"Stop Moudle Error:{module_id},{str(err)} - {traceback.format_exc()}", exc_info=True) + try: + module.stop() + logger.debug(f"Moudle Stoped:{module_id}") + except Exception as err: + logger.error(f"Stop Moudle Error:{module_id},{str(err)} - {traceback.format_exc()}", exc_info=True) logger.info("所有模块停止完成") def reload(self): diff --git a/app/modules/__init__.py b/app/modules/__init__.py index a82110f39..5b98ee1b3 100644 --- a/app/modules/__init__.py +++ b/app/modules/__init__.py @@ -1,8 +1,10 @@ +import threading from abc import abstractmethod, ABCMeta from typing import Generic, Tuple, Union, TypeVar, Type, Dict, Optional, Callable from pathlib import Path from app.helper.service import ServiceConfigHelper +from app.log import logger from app.schemas import Notification, NotificationConf, MediaServerConf, DownloaderConf from app.schemas.types import ModuleType, DownloaderType, MediaServerType, MessageChannel, StorageSchema, \ OtherModulesType, SystemConfigKey @@ -15,8 +17,21 @@ class _ModuleBase(ConfigReloadMixin, metaclass=ABCMeta): 输入参数与输出参数一致的,或没有输出的,可以被多个模块重复实现 """ - def on_config_changed(self): - self.init_module() + def __init__(self) -> None: + """初始化模块生命周期锁""" + super().__init__() + self._reload_lock = threading.RLock() + + def on_config_changed(self) -> None: + """串行停止旧资源并按最新配置重新初始化模块""" + with self._reload_lock: + try: + self.stop() + except Exception as err: + logger.error( + f"停止 {self.get_reload_name()} 旧资源失败,继续按最新配置初始化:{err}" + ) + self.init_module() def get_reload_name(self): return self.get_name() diff --git a/app/modules/discord/__init__.py b/app/modules/discord/__init__.py index 1d3b5efcc..a3bd73428 100644 --- a/app/modules/discord/__init__.py +++ b/app/modules/discord/__init__.py @@ -58,7 +58,6 @@ class DiscordModule(_ModuleBase, _MessageBase[Discord]): if not Discord: logger.error("Discord 依赖未就绪(需要安装 discord.py==2.6.4),模块未启动") return - self.stop() super().init_service( service_name=Discord.__name__.lower(), service_type=Discord ) @@ -89,12 +88,13 @@ class DiscordModule(_ModuleBase, _MessageBase[Discord]): """ return 4 - def stop(self): - """ - 停止模块 - """ + def stop(self) -> None: + """停止模块""" for client in self.get_instances().values(): - client.stop() + try: + client.stop() + except Exception as err: + logger.error(f"停止Discord模块实例失败:{err}") def test(self) -> Optional[Tuple[bool, str]]: """ diff --git a/app/modules/feishu/__init__.py b/app/modules/feishu/__init__.py index 660942c57..8a0d8fa3c 100644 --- a/app/modules/feishu/__init__.py +++ b/app/modules/feishu/__init__.py @@ -10,7 +10,6 @@ from app.schemas.types import ModuleType class FeishuModule(_ModuleBase, _MessageBase[Feishu]): def init_module(self) -> None: - self.stop() super().init_service(service_name=Feishu.__name__.lower(), service_type=Feishu) self._channel = MessageChannel.Feishu @@ -30,13 +29,13 @@ class FeishuModule(_ModuleBase, _MessageBase[Feishu]): def get_priority() -> int: return 2 - def stop(self): + def stop(self) -> None: + """停止模块""" for client in self.get_instances().values(): - if hasattr(client, "stop"): - try: - client.stop() - except Exception as err: - logger.error(f"停止飞书模块实例失败:{err}") + try: + client.stop() + except Exception as err: + logger.error(f"停止飞书模块实例失败:{err}") def test(self) -> Optional[Tuple[bool, str]]: if not self.get_instances(): diff --git a/app/modules/filter/__init__.py b/app/modules/filter/__init__.py index a47cab68e..59b5f81cf 100644 --- a/app/modules/filter/__init__.py +++ b/app/modules/filter/__init__.py @@ -92,13 +92,6 @@ class FilterModule(_ModuleBase): self.rule_set = deepcopy(self.builtin_rule_set) self.__init_custom_rules() - def on_config_changed(self) -> None: - """ - 自定义过滤或 Meta 识别配置变更后重建规则集并刷新 Rust Meta 配置缓存。 - """ - clear_rust_parse_options_cache() - self.init_module() - def __init_custom_rules(self): """ 加载用户自定义规则,如跟内置规则冲突,以用户自定义规则为准 @@ -137,10 +130,8 @@ class FilterModule(_ModuleBase): return 4 def stop(self) -> None: - """ - 停止过滤器模块。 - """ - pass + """停止模块""" + clear_rust_parse_options_cache() def test(self) -> None: """ diff --git a/app/modules/plex/__init__.py b/app/modules/plex/__init__.py index ddb83e5fe..f386658f6 100644 --- a/app/modules/plex/__init__.py +++ b/app/modules/plex/__init__.py @@ -44,13 +44,14 @@ class PlexModule(_ModuleBase, _MediaServerBase[Plex]): """ return 3 - def stop(self): - """ - 停止模块服务 - """ + def stop(self) -> None: + """停止模块""" for server in self.get_instances().values(): - if server: - server.close() + try: + if server: + server.close() + except Exception as err: + logger.error(f"停止Plex模块实例失败:{err}") def test(self) -> Optional[Tuple[bool, str]]: """ diff --git a/app/modules/qqbot/__init__.py b/app/modules/qqbot/__init__.py index 1016e8a12..5a5935b94 100644 --- a/app/modules/qqbot/__init__.py +++ b/app/modules/qqbot/__init__.py @@ -46,7 +46,6 @@ class QQBotModule(_ModuleBase, _MessageBase[QQBot]): ) def init_module(self) -> None: - self.stop() super().init_service(service_name=QQBot.__name__.lower(), service_type=QQBot) self._channel = MessageChannel.QQ @@ -67,9 +66,12 @@ class QQBotModule(_ModuleBase, _MessageBase[QQBot]): return 10 def stop(self) -> None: + """停止模块""" for client in self.get_instances().values(): - if hasattr(client, "stop"): + try: client.stop() + except Exception as err: + logger.error(f"停止QQ Bot模块实例失败:{err}") def test(self) -> Optional[Tuple[bool, str]]: if not self.get_instances(): diff --git a/app/modules/slack/__init__.py b/app/modules/slack/__init__.py index f5a00ae48..06f2ba56e 100644 --- a/app/modules/slack/__init__.py +++ b/app/modules/slack/__init__.py @@ -69,12 +69,13 @@ class SlackModule(_ModuleBase, _MessageBase[Slack]): """ return 3 - def stop(self): - """ - 停止模块 - """ + def stop(self) -> None: + """停止模块""" for client in self.get_instances().values(): - client.stop() + try: + client.stop() + except Exception as err: + logger.error(f"停止Slack模块实例失败:{err}") def test(self) -> Optional[Tuple[bool, str]]: """ diff --git a/app/modules/telegram/__init__.py b/app/modules/telegram/__init__.py index 912e174ea..13b701132 100644 --- a/app/modules/telegram/__init__.py +++ b/app/modules/telegram/__init__.py @@ -62,12 +62,13 @@ class TelegramModule(_ModuleBase, _MessageBase[Telegram]): """ return 0 - def stop(self): - """ - 停止模块 - """ + def stop(self) -> None: + """停止模块""" for client in self.get_instances().values(): - client.stop() + try: + client.stop() + except Exception as err: + logger.error(f"停止Telegram模块实例失败:{err}") def test(self) -> Optional[Tuple[bool, str]]: """ diff --git a/app/modules/telegram/telegram.py b/app/modules/telegram/telegram.py index 5eefb533d..650fb2147 100644 --- a/app/modules/telegram/telegram.py +++ b/app/modules/telegram/telegram.py @@ -1536,14 +1536,19 @@ class Telegram: # 清理菜单命令 self._bot.delete_my_commands() - def stop(self): + def stop(self) -> None: """ 停止Telegram消息接收服务 """ # 停止所有typing任务 for chat_id in list(self._typing_tasks.keys()): self._stop_typing_task(chat_id) - if self._bot: - self._bot.stop_polling() + if not self._bot: + return + + self._bot.stop_bot() + if self._polling_thread: self._polling_thread.join() - logger.info("Telegram消息接收服务已停止") + self._polling_thread = None + self._bot = None + logger.info("Telegram消息接收服务已停止") diff --git a/app/modules/themoviedb/__init__.py b/app/modules/themoviedb/__init__.py index 9aab94b95..966b2b75b 100644 --- a/app/modules/themoviedb/__init__.py +++ b/app/modules/themoviedb/__init__.py @@ -43,12 +43,6 @@ class TheMovieDbModule(_ModuleBase): self.category = CategoryHelper() self.scraper = TmdbScraper() - def on_config_changed(self): - # 停止模块 - self.stop() - # 初始化模块 - self.init_module() - @staticmethod def get_name() -> str: return "TheMovieDb" @@ -74,9 +68,13 @@ class TheMovieDbModule(_ModuleBase): """ return 1 - def stop(self): - self.cache.save() - self.tmdb.close() + def stop(self) -> None: + """停止模块""" + # 缓存持久化失败不能阻断 HTTP 客户端关闭 + try: + self.cache.save() + finally: + self.tmdb.close() def test(self) -> Tuple[bool, str]: """ diff --git a/app/modules/trimemedia/__init__.py b/app/modules/trimemedia/__init__.py index aa2d25cba..616ee02c7 100644 --- a/app/modules/trimemedia/__init__.py +++ b/app/modules/trimemedia/__init__.py @@ -61,10 +61,14 @@ class TrimeMediaModule(_ModuleBase, _MediaServerBase[TrimeMedia]): logger.info(f"飞牛影视 {name} 连接断开,尝试重连 ...") server.reconnect() - def stop(self): + def stop(self) -> None: + """停止模块""" for server in self.get_instances().values(): - if server.is_authenticated(): - server.disconnect() + try: + if server.is_authenticated(): + server.disconnect() + except Exception as err: + logger.error(f"停止飞牛影视模块实例失败:{err}") def test(self) -> Optional[Tuple[bool, str]]: """ diff --git a/app/modules/ugreen/__init__.py b/app/modules/ugreen/__init__.py index 705ece72b..3fc342fb1 100644 --- a/app/modules/ugreen/__init__.py +++ b/app/modules/ugreen/__init__.py @@ -60,10 +60,14 @@ class UgreenModule(_ModuleBase, _MediaServerBase[Ugreen]): logger.info(f"绿联影视 {name} 连接断开,尝试重连 ...") server.reconnect() - def stop(self): + def stop(self) -> None: + """停止模块""" for server in self.get_instances().values(): - if server.is_authenticated(): - server.disconnect() + try: + if server.is_authenticated(): + server.disconnect() + except Exception as err: + logger.error(f"停止绿联影视模块实例失败:{err}") def test(self) -> Optional[Tuple[bool, str]]: """ diff --git a/app/modules/wechat/__init__.py b/app/modules/wechat/__init__.py index 580a00870..e1e4e5e71 100644 --- a/app/modules/wechat/__init__.py +++ b/app/modules/wechat/__init__.py @@ -24,7 +24,6 @@ class WechatModule(_ModuleBase, _MessageBase[WeChat]): """ 初始化模块 """ - self.stop() super().init_service(service_name=WeChat.__name__.lower(), service_type=self._create_client) self._channel = MessageChannel.Wechat @@ -54,13 +53,14 @@ class WechatModule(_ModuleBase, _MessageBase[WeChat]): """ return 1 - def stop(self): + def stop(self) -> None: + """停止模块""" for client in self.get_instances().values(): - if hasattr(client, "stop"): - try: + try: + if hasattr(client, "stop"): client.stop() - except Exception as err: - logger.error(f"停止微信模块实例失败:{err}") + except Exception as err: + logger.error(f"停止微信模块实例失败:{err}") @staticmethod def _is_bot_mode(config: dict) -> bool: diff --git a/app/modules/wechatclawbot/__init__.py b/app/modules/wechatclawbot/__init__.py index 5055ebe89..040607423 100644 --- a/app/modules/wechatclawbot/__init__.py +++ b/app/modules/wechatclawbot/__init__.py @@ -23,7 +23,6 @@ class WechatClawBotModule(_ModuleBase, _MessageBase[WechatClawBot]): def init_module(self) -> None: """初始化模块。""" - self.stop() super().init_service( service_name=WechatClawBot.__name__.lower(), service_type=WechatClawBot ) @@ -49,14 +48,13 @@ class WechatClawBotModule(_ModuleBase, _MessageBase[WechatClawBot]): """获取模块优先级。""" return 2 - def stop(self): - """停止模块。""" + def stop(self) -> None: + """停止模块""" for client in self.get_instances().values(): - if hasattr(client, "stop"): - try: - client.stop() - except Exception as err: - logger.error(f"停止微信 ClawBot 模块实例失败:{err}") + try: + client.stop() + except Exception as err: + logger.error(f"停止微信 ClawBot 模块实例失败:{err}") def test(self) -> Optional[Tuple[bool, str]]: """测试模块连接性。""" diff --git a/tests/test_module_lifecycle.py b/tests/test_module_lifecycle.py new file mode 100644 index 000000000..597bbc04f --- /dev/null +++ b/tests/test_module_lifecycle.py @@ -0,0 +1,187 @@ +import threading +from unittest.mock import Mock, patch + +import pytest + +from app.modules import _MessageBase +from app.modules.discord import DiscordModule +from app.modules.feishu import FeishuModule +from app.modules.filter import FilterModule +from app.modules.plex import PlexModule +from app.modules.qqbot import QQBotModule +from app.modules.slack import SlackModule +from app.modules.telegram import TelegramModule +from app.modules.telegram.telegram import Telegram +from app.modules.themoviedb import TheMovieDbModule +from app.modules.trimemedia import TrimeMediaModule +from app.modules.ugreen import UgreenModule +from app.modules.wechat import WechatModule +from app.modules.wechatclawbot import WechatClawBotModule + + +def test_config_reload_stops_before_initializing_latest_generation(): + """同一模块的重载必须串行,并依次停止和初始化 generation。""" + module = TelegramModule() + call_order = [] + reload_started = threading.Event() + reload_finished = threading.Event() + + def reload_module(): + reload_started.set() + module.on_config_changed() + reload_finished.set() + + with patch.object( + module, "stop", side_effect=lambda: call_order.append("stop") + ), patch.object( + _MessageBase, + "init_service", + side_effect=lambda **_kwargs: call_order.append("init"), + ): + module._reload_lock.acquire() + try: + reload_thread = threading.Thread(target=reload_module) + reload_thread.start() + assert reload_started.wait(1) + assert not reload_finished.wait(0.1) + finally: + module._reload_lock.release() + + assert reload_finished.wait(1) + reload_thread.join() + + assert call_order == ["stop", "init"] + + +def test_initialization_does_not_stop_a_fresh_module_generation(): + """首次初始化只创建资源,停止旧 generation 由重载入口负责。""" + module = TelegramModule() + + with patch.object(module, "stop") as stop, patch.object( + _MessageBase, "init_service" + ) as init_service: + module.init_module() + + stop.assert_not_called() + init_service.assert_called_once() + + +def test_config_reload_initializes_latest_generation_after_stop_failure(): + """旧资源停止异常只记录错误,不阻止最新配置完成初始化。""" + module = TelegramModule() + call_order = [] + + def stop_with_failure(): + call_order.append("stop") + raise RuntimeError("stop failed") + + with patch.object( + module, + "stop", + side_effect=stop_with_failure, + ), patch.object( + _MessageBase, + "init_service", + side_effect=lambda **_kwargs: call_order.append("init"), + ): + module.on_config_changed() + + assert call_order == ["stop", "init"] + + +def test_tmdb_reload_closes_old_client_when_cache_save_fails(): + """TMDB 缓存保存失败时仍须关闭旧客户端并初始化最新配置。""" + module = TheMovieDbModule() + module.cache = Mock() + module.cache.save.side_effect = OSError("cache write failed") + module.tmdb = Mock() + + with patch.object(module, "init_module") as init_module: + module.on_config_changed() + + module.tmdb.close.assert_called_once_with() + init_module.assert_called_once_with() + + +def test_filter_reload_uses_shared_module_lifecycle_lock(): + """过滤规则重载必须经过模块基类的串行 stop 和 init。""" + module = FilterModule() + reload_started = threading.Event() + reload_finished = threading.Event() + call_order = [] + + def reload_module(): + reload_started.set() + module.on_config_changed() + reload_finished.set() + + with patch( + "app.modules.filter.clear_rust_parse_options_cache", + side_effect=lambda: call_order.append("stop"), + ), patch.object( + module, "init_module", side_effect=lambda: call_order.append("init") + ): + module._reload_lock.acquire() + try: + reload_thread = threading.Thread(target=reload_module) + reload_thread.start() + assert reload_started.wait(1) + assert not reload_finished.wait(0.1) + finally: + module._reload_lock.release() + + assert reload_finished.wait(1) + reload_thread.join() + + assert call_order == ["stop", "init"] + + +@pytest.mark.parametrize( + ("module_type", "stop_method", "requires_authentication"), + [ + (DiscordModule, "stop", False), + (FeishuModule, "stop", False), + (QQBotModule, "stop", False), + (SlackModule, "stop", False), + (TelegramModule, "stop", False), + (WechatModule, "stop", False), + (WechatClawBotModule, "stop", False), + (PlexModule, "close", False), + (TrimeMediaModule, "disconnect", True), + (UgreenModule, "disconnect", True), + ], +) +def test_module_stop_isolates_each_service_instance( + module_type, stop_method, requires_authentication +): + """单个服务停止失败时必须继续关闭同模块的其余实例。""" + module = module_type() + failed_client = Mock() + healthy_client = Mock() + getattr(failed_client, stop_method).side_effect = RuntimeError("stop failed") + if requires_authentication: + failed_client.is_authenticated.return_value = True + healthy_client.is_authenticated.return_value = True + module._instances = {"failed": failed_client, "healthy": healthy_client} + + module.stop() + + getattr(failed_client, stop_method).assert_called_once_with() + getattr(healthy_client, stop_method).assert_called_once_with() + + +def test_telegram_stop_closes_sdk_and_waits_for_polling_thread(): + """客户端停止完成后不得保留 SDK worker 或 polling 线程句柄。""" + client = Telegram.__new__(Telegram) + bot = Mock() + client._bot = bot + polling_thread = Mock() + client._polling_thread = polling_thread + + client.stop() + client.stop() + + bot.stop_bot.assert_called_once_with() + polling_thread.join.assert_called_once_with() + assert client._bot is None + assert client._polling_thread is None