diff --git a/app/agent/__init__.py b/app/agent/__init__.py index b349875ba..fb30b23a0 100644 --- a/app/agent/__init__.py +++ b/app/agent/__init__.py @@ -34,6 +34,7 @@ from app.log import logger from app.schemas import Notification, NotificationType from app.schemas.message import ChannelCapabilityManager, ChannelCapability from app.schemas.types import MessageChannel +from app.utils.identity import SYSTEM_INTERNAL_USER_ID class AgentChain(ChainBase): @@ -543,16 +544,12 @@ class MoviePilotAgent: """ 通过原渠道发送消息给用户 """ - user_id = self.user_id - if self.user_id == "system": - user_id = None - await AgentChain().async_post_message( Notification( channel=self.channel, source=self.source, mtype=NotificationType.Agent, - userid=user_id, + userid=self.user_id, username=self.username, title=title, text=message, @@ -853,7 +850,7 @@ class AgentManager: try: # 每次使用唯一的 session_id，避免共享上下文 session_id = f"__agent_heartbeat_{uuid.uuid4().hex[:12]}__" - user_id = "system" + user_id = SYSTEM_INTERNAL_USER_ID logger.info("智能体心跳唤醒：开始检查待处理任务...") @@ -948,7 +945,7 @@ class AgentManager: return session_id = f"__agent_retry_transfer_batch_{uuid.uuid4().hex[:8]}__" - user_id = "system" + user_id = SYSTEM_INTERNAL_USER_ID ids_str = ", ".join(str(i) for i in history_ids) logger.info( @@ -1107,7 +1104,7 @@ class AgentManager: 手动触发单条历史记录的 AI 整理。 """ session_id = f"__agent_manual_redo_{history_id}_{uuid.uuid4().hex[:8]}__" - user_id = "system" + user_id = SYSTEM_INTERNAL_USER_ID agent = MoviePilotAgent( session_id=session_id, user_id=user_id, diff --git a/app/chain/__init__.py b/app/chain/__init__.py index ec39fe412..2b9f54c51 100644 --- a/app/chain/__init__.py +++ b/app/chain/__init__.py @@ -38,6 +38,7 @@ from app.schemas import ( TransferDirectoryConf, MessageResponse, ) +from app.utils.identity import normalize_internal_user_id from app.schemas.category import CategoryConfig from app.schemas.types import ( TorrentStatus, @@ -119,6 +120,21 @@ class ChainBase(metaclass=ABCMeta): """ self.filecache.delete(filename) + @staticmethod + def _normalize_notification_for_dispatch( + message: Notification + ) -> Notification: + """ + 规范化待发送的通知消息。 + 后台任务会复用内部占位用户ID作为会话身份，这里在真正发送前清空， + 让消息重新走默认通知路由或基于 targets 的目标解析。 + """ + dispatch_message = copy.deepcopy(message) + dispatch_message.userid = normalize_internal_user_id( + dispatch_message.userid + ) + return dispatch_message + async def async_remove_cache(self, filename: str) -> None: """ 异步删除缓存，同时删除Redis和本地缓存 @@ -1119,10 +1135,13 @@ class ChainBase(metaclass=ABCMeta): # 保存消息 self.messagehelper.put(message, role="user", title=message.title) self.messageoper.add(**message.model_dump()) + dispatch_message = self._normalize_notification_for_dispatch(message) # 发送消息按设置隔离 - if not message.userid and message.mtype: + if not dispatch_message.userid and dispatch_message.mtype: # 消息隔离设置 - notify_action = ServiceConfigHelper.get_notification_switch(message.mtype) + notify_action = ServiceConfigHelper.get_notification_switch( + dispatch_message.mtype + ) if notify_action: # 'admin' 'user,admin' 'user' 'all' actions = notify_action.split(",") @@ -1131,7 +1150,7 @@ class ChainBase(metaclass=ABCMeta): send_orignal = False useroper = UserOper() for action in actions: - send_message = copy.deepcopy(message) + send_message = copy.deepcopy(dispatch_message) if action == "admin" and not admin_sended: # 仅发送管理员 logger.info(f"{send_message.mtype} 的消息已设置发送给管理员") @@ -1186,13 +1205,13 @@ class ChainBase(metaclass=ABCMeta): # 发送消息事件 self.eventmanager.send_event( etype=EventType.NoticeMessage, - data={**message.model_dump(), "type": message.mtype}, + data={**dispatch_message.model_dump(), "type": dispatch_message.mtype}, ) # 按原消息发送 self.messagequeue.send_message( "post_message", - message=message, - immediately=True if message.userid else False, + message=dispatch_message, + immediately=True if dispatch_message.userid else False, **kwargs, ) @@ -1233,10 +1252,13 @@ class ChainBase(metaclass=ABCMeta): # 保存消息 self.messagehelper.put(message, role="user", title=message.title) await self.messageoper.async_add(**message.model_dump()) + dispatch_message = self._normalize_notification_for_dispatch(message) # 发送消息按设置隔离 - if not message.userid and message.mtype: + if not dispatch_message.userid and dispatch_message.mtype: # 消息隔离设置 - notify_action = ServiceConfigHelper.get_notification_switch(message.mtype) + notify_action = ServiceConfigHelper.get_notification_switch( + dispatch_message.mtype + ) if notify_action: # 'admin' 'user,admin' 'user' 'all' actions = notify_action.split(",") @@ -1245,7 +1267,7 @@ class ChainBase(metaclass=ABCMeta): send_orignal = False useroper = UserOper() for action in actions: - send_message = copy.deepcopy(message) + send_message = copy.deepcopy(dispatch_message) if action == "admin" and not admin_sended: # 仅发送管理员 logger.info(f"{send_message.mtype} 的消息已设置发送给管理员") @@ -1300,13 +1322,13 @@ class ChainBase(metaclass=ABCMeta): # 发送消息事件 await self.eventmanager.async_send_event( etype=EventType.NoticeMessage, - data={**message.model_dump(), "type": message.mtype}, + data={**dispatch_message.model_dump(), "type": dispatch_message.mtype}, ) # 按原消息发送 await self.messagequeue.async_send_message( "post_message", - message=message, - immediately=True if message.userid else False, + message=dispatch_message, + immediately=True if dispatch_message.userid else False, **kwargs, ) @@ -1324,11 +1346,12 @@ class ChainBase(metaclass=ABCMeta): message, role="user", note=note_list, title=message.title ) self.messageoper.add(**message.model_dump(), note=note_list) + dispatch_message = self._normalize_notification_for_dispatch(message) return self.messagequeue.send_message( "post_medias_message", - message=message, + message=dispatch_message, medias=medias, - immediately=True if message.userid else False, + immediately=True if dispatch_message.userid else False, ) def post_torrents_message( @@ -1345,11 +1368,12 @@ class ChainBase(metaclass=ABCMeta): message, role="user", note=note_list, title=message.title ) self.messageoper.add(**message.model_dump(), note=note_list) + dispatch_message = self._normalize_notification_for_dispatch(message) return self.messagequeue.send_message( "post_torrents_message", - message=message, + message=dispatch_message, torrents=torrents, - immediately=True if message.userid else False, + immediately=True if dispatch_message.userid else False, ) def delete_message( @@ -1411,7 +1435,10 @@ class ChainBase(metaclass=ABCMeta): :param message: 消息体 :return: 消息响应（包含message_id, chat_id等） """ - return self.run_module("send_direct_message", message=message) + return self.run_module( + "send_direct_message", + message=self._normalize_notification_for_dispatch(message), + ) def metadata_img( self, diff --git a/app/utils/identity.py b/app/utils/identity.py new file mode 100644 index 000000000..fee8094df --- /dev/null +++ b/app/utils/identity.py @@ -0,0 +1,27 @@ +from typing import Optional, Union + +# 后台任务会话使用的内部占位用户ID。 +# 它只用于在 agent/memory/session 侧标识“系统触发的任务”， +# 不能直接作为真实消息接收人下发到 Telegram/企业微信 等通知渠道。 +SYSTEM_INTERNAL_USER_ID = "system" + + +def is_internal_user_id(userid: Optional[Union[str, int]]) -> bool: + """ + 判断是否为系统内部占位用户ID。 + """ + return ( + isinstance(userid, str) + and userid.strip().lower() == SYSTEM_INTERNAL_USER_ID + ) + + +def normalize_internal_user_id( + userid: Optional[Union[str, int]] +) -> Optional[Union[str, int]]: + """ + 将系统内部占位用户ID归一化为 None，避免被通知渠道误认为真实接收人。 + """ + if is_internal_user_id(userid): + return None + return userid diff --git a/tests/test_system_notification_dispatch.py b/tests/test_system_notification_dispatch.py new file mode 100644 index 000000000..e14d6297e --- /dev/null +++ b/tests/test_system_notification_dispatch.py @@ -0,0 +1,70 @@ +import sys +import unittest +from types import ModuleType +from unittest.mock import patch + +sys.modules.setdefault("qbittorrentapi", ModuleType("qbittorrentapi")) +setattr(sys.modules["qbittorrentapi"], "TorrentFilesList", list) +sys.modules.setdefault("transmission_rpc", ModuleType("transmission_rpc")) +setattr(sys.modules["transmission_rpc"], "File", object) +sys.modules.setdefault("psutil", ModuleType("psutil")) + +from app.chain.message import MessageChain +from app.schemas import Notification +from app.utils.identity import ( + SYSTEM_INTERNAL_USER_ID, + is_internal_user_id, + normalize_internal_user_id, +) + + +class TestSystemNotificationDispatch(unittest.TestCase): + def test_internal_userid_identity_helpers(self): + self.assertTrue(is_internal_user_id(SYSTEM_INTERNAL_USER_ID)) + self.assertTrue(is_internal_user_id(" System ")) + self.assertIsNone(normalize_internal_user_id(SYSTEM_INTERNAL_USER_ID)) + self.assertEqual(normalize_internal_user_id("10001"), "10001") + + def test_post_message_normalizes_internal_userid_before_queueing(self): + chain = MessageChain() + message = Notification( + userid=SYSTEM_INTERNAL_USER_ID, + username="admin", + title="后台报告", + text="任务完成", + ) + + with patch("app.chain.MessageTemplateHelper.render", return_value=message), patch.object( + chain.messagehelper, "put" + ), patch.object(chain.messageoper, "add"), patch.object( + chain.eventmanager, "send_event" + ) as send_event, patch.object( + chain.messagequeue, "send_message" + ) as send_message: + chain.post_message(message) + + event_payload = send_event.call_args.kwargs["data"] + queued_message = send_message.call_args.kwargs["message"] + + self.assertIsNone(event_payload["userid"]) + self.assertIsNone(queued_message.userid) + self.assertFalse(send_message.call_args.kwargs["immediately"]) + + def test_send_direct_message_normalizes_internal_userid(self): + chain = MessageChain() + message = Notification( + userid=SYSTEM_INTERNAL_USER_ID, + username="admin", + title="后台报告", + text="任务完成", + ) + + with patch.object(chain, "run_module") as run_module: + chain.send_direct_message(message) + + sent_message = run_module.call_args.kwargs["message"] + self.assertIsNone(sent_message.userid) + + +if __name__ == "__main__": + unittest.main()