fix(notification): mark realtime notification type and severity

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

app/application/messaging/message.py

@@ -28,7 +28,7 @@ from app.runtime.stop import runtime_stop_state
 from app.schemas.message import Message
 from app.schemas.tmdb import TmdbEpisode
 from app.schemas.transfer import TransferInfo
-from app.schemas.types import MUSIC_ENTITY_ALBUM, SystemConfigKey
+from app.schemas.types import MUSIC_ENTITY_ALBUM, MessageType, SystemConfigKey
 
 # 专辑名尾部的括号年份标记；重命名模板会独立追加 `({{year}})`，
 # 标签或目录名中自带的尾部年份若不剥离，会生成重复年份的目录名（issue #6355）
@@ -1189,6 +1189,10 @@ class MessageQueueManager(metaclass=SingletonClass):
         return True
 
 
+# 实时通知严重级别：前端据此决定图标与颜色，error/warning 优先于业务类型展示
+NotificationLevel = Literal["info", "success", "warning", "error"]
+
+
 class MessageHelper(metaclass=Singleton):
     """
     消息队列管理器，负责系统和插件实时消息的 SSE 推送
@@ -1234,13 +1238,24 @@ class MessageHelper(metaclass=Singleton):
         self._recent_notification_keys.set(key, True)
         return False
 
-    def put(self, message: Any, role: str = "plugin", title: str = None, note: Union[list, dict] = None):
+    def put(
+            self,
+            message: Any,
+            role: str = "plugin",
+            title: str = None,
+            note: Union[list, dict] = None,
+            mtype: Optional[MessageType] = None,
+            level: Optional[NotificationLevel] = None,
+    ):
         """
         存消息
         :param message: 消息
-        :param role: 消息通道 system：系统消息，plugin：插件消息
+        :param role: 消息通道 system：系统消息，plugin：插件消息；只表示来源，不表示严重程度
         :param title: 标题
         :param note: 附件json
+        :param mtype: 业务类型，前端据此选择业务图标，例如订阅消息显示订阅图标
+        :param level: 严重级别 info/success/warning/error；未设置时前端按普通通知展示，
+                      失败或需要用户处理的消息必须显式标记，才会显示为警示样式
         """
         if role not in ["system", "plugin"]:
             return
@@ -1254,7 +1269,9 @@ class MessageHelper(metaclass=Singleton):
             "title": title,
             "text": message,
             "date": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()),
-            "note": note
+            "note": note,
+            "mtype": mtype.value if mtype else None,
+            "level": level,
         }))
 
     def get(self, role: str = "system") -> Optional[str]:

app/chain/base.py

@@ -179,6 +179,7 @@ class ChainBase(RecognitionMixin, MessageProcessingMixin, NotificationMixin, met
             title=f"{plugin_name} 运行失败",
             message=f"{plugin_id}.{method}：{str(err).strip() or type(err).__name__}",
             role="plugin",
+            level="error",
         )
         self.eventmanager.send_event(
             EventType.SystemError,
@@ -203,6 +204,7 @@ class ChainBase(RecognitionMixin, MessageProcessingMixin, NotificationMixin, met
             title=f"{module_name}运行失败",
             message=f"{module_id}.{method}：{str(err).strip() or type(err).__name__}",
             role="system",
+            level="error",
         )
         self.eventmanager.send_event(
             EventType.SystemError,

app/chain/download/submission.py

@@ -443,6 +443,7 @@ class DownloadSubmissionOwner(_DownloadResourceOwner):
                     f"{media.type.value} {media.title_year} 未找到下载目录！",
                     title="下载失败",
                     role="system",
+                    level="error",
                 )
             return None, error_msg or "未找到下载目录"
         download_uri = FileURI(storage=storage, path=download_dir.as_posix()).uri

app/chain/message.py

@@ -1476,7 +1476,7 @@ class MessageChain(ChainBase):
 
         except Exception as e:
             logger.error(f"处理AI智能体消息失败: {e}", exc_info=True)
-            self.messagehelper.put("智能助手执行失败，请稍后重试", role="system", title="MoviePilot助手")
+            self.messagehelper.put("智能助手执行失败，请稍后重试", role="system", title="MoviePilot助手", level="error")
             return False
 
     def _transcribe_audio_refs(

app/chain/site.py

@@ -586,7 +586,7 @@ class SiteChain(InteractionChainMixin, ChainBase):
             if progress_callback:
                 progress_callback(value=100, text=f"CookieCloud同步失败：{msg}")
             if manual:
-                self.messagehelper.put(msg, title="CookieCloud同步失败", role="system")
+                self.messagehelper.put(msg, title="CookieCloud同步失败", role="system", level="error")
             return False, msg
         siteshelper = SitesHelper()
         repository = self.site_repository
@@ -641,7 +641,7 @@ class SiteChain(InteractionChainMixin, ChainBase):
         if fail_count > 0:
             ret_msg += f"，{fail_count}个站点添加失败，下次同步时将重试，也可以手动添加"
         if manual:
-            self.messagehelper.put(ret_msg, title="CookieCloud同步成功", role="system")
+            self.messagehelper.put(ret_msg, title="CookieCloud同步成功", role="system", level="success")
         logger.info(f"CookieCloud同步成功：{ret_msg}")
         if progress_callback:
             progress_callback(value=100, text=f"CookieCloud同步成功：{ret_msg}")

app/chain/subscribe/search.py

@@ -53,6 +53,7 @@ from app.runtime.stop import runtime_stop_state
 from app.runtime.tasks import get_task_registry
 from app.schemas.types import (
     MediaType,
+    MessageType,
     SystemConfigKey,
 )
 
@@ -897,21 +898,25 @@ class SubscribeSearchOwner(_SubscribeSearchQueueOwner):
         if not manual:
             return
         if not subscribes:
-            self.messagehelper.put("没有找到订阅！", title="订阅搜索", role="system")
+            self.messagehelper.put(
+                "没有找到订阅！", title="订阅搜索", role="system", mtype=MessageType.Subscribe, level="warning"
+            )
         elif sid:
             message = (
                 f"{subscribes[0].name} 搜索完成！"
                 if processed
                 else f"{subscribes[0].name} 已安排搜索，系统会自动继续处理。"
             )
-            self.messagehelper.put(message, title="订阅搜索", role="system")
+            self.messagehelper.put(message, title="订阅搜索", role="system", mtype=MessageType.Subscribe)
         elif sids is not None:
             for subscribe in processed:
-                self.messagehelper.put(f"{subscribe.name} 搜索完成！", title="订阅搜索", role="system")
+                self.messagehelper.put(
+                    f"{subscribe.name} 搜索完成！", title="订阅搜索", role="system", mtype=MessageType.Subscribe
+                )
         else:
             message = (
                 "所有订阅搜索完成！"
                 if len(processed) == len(subscribes)
                 else "订阅搜索已安排，暂未完成的项目会自动继续处理。"
             )
-            self.messagehelper.put(message, title="订阅搜索", role="system")
+            self.messagehelper.put(message, title="订阅搜索", role="system", mtype=MessageType.Subscribe)

app/command.py

@@ -436,7 +436,7 @@ class Command(metaclass=Singleton):
                     f"执行命令 {cmd} 出错：{str(err)} - {traceback.format_exc()}"
                 )
                 self.messagehelper.put(
-                    title=f"执行命令 {cmd} 出错", message=str(err), role="system"
+                    title=f"执行命令 {cmd} 出错", message=str(err), role="system", level="error"
                 )
 
     @staticmethod

app/scheduler/execution.py

@@ -113,6 +113,7 @@ class SchedulerExecutionOwner(_SchedulerOwnerBase):
             title=f"{job.get('name')} 执行失败",
             message=str(error),
             role="system",
+            level="error",
         )
         eventmanager.send_event(
             EventType.SystemError,

app/startup/initializers/modules.py

@@ -274,6 +274,7 @@ def notify_event_error(title: str, message: str) -> None:
         title=title,
         message=message,
         role="system",
+        level="error",
     )
 
 
@@ -489,7 +490,7 @@ def check_auth():
     """
     if SitesHelper().auth_level < 2:
         err_msg = "用户认证失败，站点相关功能将无法使用！"
-        MessageHelper().put(f"注意：{err_msg}", title="用户认证", role="system")
+        MessageHelper().put(f"注意：{err_msg}", title="用户认证", role="system", level="warning")
         CommandChain().post_message(
             Message(
                 mtype=MessageType.Manual,

tests/test_chain_runtime_context.py

@@ -120,6 +120,7 @@ def test_module_error_notification_preserves_reason(source, asynchronous, error,
         title="测试插件 运行失败" if source == "plugin" else "测试模块运行失败",
         message=f"{source_id}.{method}：{expected_reason}",
         role=source,
+        level="error",
     )
     chain.eventmanager.send_event.assert_called_once()
     event_type, payload = chain.eventmanager.send_event.call_args.args

tests/test_message_notifications.py

@@ -132,6 +132,24 @@ def test_system_helper_message_only_enters_sse_queue() -> None:
     assert realtime_message["text"] == "调度任务执行失败"
 
 
+def test_system_helper_message_carries_business_type_and_level() -> None:
+    """
+    实时消息应携带业务类型和严重级别，未标记级别的系统消息不应被当作错误。
+    """
+    helper = MessageHelper()
+    _reset_message_helper(helper)
+
+    helper.put("测试剧集 搜索完成！", role="system", title="订阅搜索", mtype=MessageType.Subscribe)
+    helper.put("用户认证失败", role="system", title="用户认证", level="warning")
+
+    subscribe_message = json.loads(helper.get())
+    assert subscribe_message["mtype"] == "订阅"
+    assert subscribe_message["level"] is None
+    warning_message = json.loads(helper.get())
+    assert warning_message["mtype"] is None
+    assert warning_message["level"] == "warning"
+
+
 def test_plugin_helper_message_deduplicates_recent_sse_messages() -> None:
     """
     短时间内相同插件实时消息只应推送一次，不写入通知历史。