feat(notification): redesign system to support multiple providers

- Add NotificationProvider base class with send() and test() methods
- Add NotificationManager for handling multiple providers simultaneously
- Add new providers: Discord, Gotify, Pushover, generic Webhook
- Migrate existing providers (Telegram, Bark, Server Chan, WeChat Work) to new architecture
- Add API endpoints for testing providers (/notification/test, /notification/test-config)
- Auto-migrate legacy single-provider configs to new multi-provider format
- Update WebUI with card-based multi-provider settings UI
- Add test button for each provider in settings
- Generic webhook supports template variables: {{title}}, {{season}}, {{episode}}, {{poster_url}}

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

Author: EstrellaXD

backend/src/module/api/__init__.py

@@ -10,6 +10,7 @@
 from .rss import router as rss_router
 from .search import router as search_router
 from .setup import router as setup_router
+from .notification import router as notification_router
 
 __all__ = "v1"
 
@@ -25,3 +26,4 @@
 v1.include_router(rss_router)
 v1.include_router(search_router)
 v1.include_router(setup_router)
+v1.include_router(notification_router)

backend/src/module/api/notification.py

@@ -0,0 +1,117 @@
+"""Notification API endpoints."""
+
+import logging
+from typing import Optional
+
+from fastapi import APIRouter
+from pydantic import BaseModel, Field
+
+from module.notification import NotificationManager
+from module.models.config import NotificationProvider as ProviderConfig
+
+logger = logging.getLogger(__name__)
+router = APIRouter(prefix="/notification", tags=["notification"])
+
+
+class TestProviderRequest(BaseModel):
+    """Request body for testing a saved provider by index."""
+
+    provider_index: int = Field(..., description="Index of the provider to test")
+
+
+class TestProviderConfigRequest(BaseModel):
+    """Request body for testing an unsaved provider configuration."""
+
+    type: str = Field(..., description="Provider type")
+    enabled: bool = Field(True, description="Whether provider is enabled")
+    token: Optional[str] = Field(None, description="Auth token")
+    chat_id: Optional[str] = Field(None, description="Chat/channel ID")
+    webhook_url: Optional[str] = Field(None, description="Webhook URL")
+    server_url: Optional[str] = Field(None, description="Server URL")
+    device_key: Optional[str] = Field(None, description="Device key")
+    user_key: Optional[str] = Field(None, description="User key")
+    api_token: Optional[str] = Field(None, description="API token")
+    template: Optional[str] = Field(None, description="Custom template")
+    url: Optional[str] = Field(None, description="URL for generic webhook")
+
+
+class TestResponse(BaseModel):
+    """Response for test notification endpoints."""
+
+    success: bool
+    message: str
+    message_zh: str = ""
+    message_en: str = ""
+
+
+@router.post("/test", response_model=TestResponse)
+async def test_provider(request: TestProviderRequest):
+    """Test a configured notification provider by its index.
+
+    Sends a test notification using the provider at the specified index
+    in the current configuration.
+    """
+    try:
+        manager = NotificationManager()
+        if request.provider_index >= len(manager):
+            return TestResponse(
+                success=False,
+                message=f"Invalid provider index: {request.provider_index}",
+                message_zh=f"无效的提供者索引: {request.provider_index}",
+                message_en=f"Invalid provider index: {request.provider_index}",
+            )
+
+        success, message = await manager.test_provider(request.provider_index)
+        return TestResponse(
+            success=success,
+            message=message,
+            message_zh="测试成功" if success else f"测试失败: {message}",
+            message_en="Test successful" if success else f"Test failed: {message}",
+        )
+    except Exception as e:
+        logger.error(f"Failed to test provider: {e}")
+        return TestResponse(
+            success=False,
+            message=str(e),
+            message_zh=f"测试失败: {e}",
+            message_en=f"Test failed: {e}",
+        )
+
+
+@router.post("/test-config", response_model=TestResponse)
+async def test_provider_config(request: TestProviderConfigRequest):
+    """Test an unsaved notification provider configuration.
+
+    Useful for testing a provider before saving it to the configuration.
+    """
+    try:
+        # Convert request to ProviderConfig
+        config = ProviderConfig(
+            type=request.type,
+            enabled=request.enabled,
+            token=request.token or "",
+            chat_id=request.chat_id or "",
+            webhook_url=request.webhook_url or "",
+            server_url=request.server_url or "",
+            device_key=request.device_key or "",
+            user_key=request.user_key or "",
+            api_token=request.api_token or "",
+            template=request.template,
+            url=request.url or "",
+        )
+
+        success, message = await NotificationManager.test_provider_config(config)
+        return TestResponse(
+            success=success,
+            message=message,
+            message_zh="测试成功" if success else f"测试失败: {message}",
+            message_en="Test successful" if success else f"Test failed: {message}",
+        )
+    except Exception as e:
+        logger.error(f"Failed to test provider config: {e}")
+        return TestResponse(
+            success=False,
+            message=str(e),
+            message_zh=f"测试失败: {e}",
+            message_en=f"Test failed: {e}",
+        )

backend/src/module/api/setup.py

@@ -8,7 +8,8 @@
 from module.conf import VERSION, settings
 from module.models import Config, ResponseModel
 from module.network import RequestContent
-from module.notification.notification import getClient
+from module.notification import PROVIDER_REGISTRY
+from module.models.config import NotificationProvider as ProviderConfig
 from module.security.jwt import get_password_hash
 
 logger = logging.getLogger(__name__)
@@ -202,39 +203,36 @@ async def test_notification(req: TestNotificationRequest):
     """Send a test notification."""
     _require_setup_needed()
 
-    NotifierClass = getClient(req.type)
-    if NotifierClass is None:
+    provider_cls = PROVIDER_REGISTRY.get(req.type.lower())
+    if provider_cls is None:
         return TestResultResponse(
             success=False,
             message_en=f"Unknown notification type: {req.type}",
             message_zh=f"未知的通知类型：{req.type}",
         )
 
     try:
-        notifier = NotifierClass(token=req.token, chat_id=req.chat_id)
-        async with notifier:
-            # Send a simple test message
-            data = {"chat_id": req.chat_id, "text": "AutoBangumi 通知测试成功！"}
-            if req.type.lower() == "telegram":
-                resp = await notifier.post_data(notifier.message_url, data)
-                if resp.status_code == 200:
-                    return TestResultResponse(
-                        success=True,
-                        message_en="Test notification sent successfully.",
-                        message_zh="测试通知发送成功。",
-                    )
-                else:
-                    return TestResultResponse(
-                        success=False,
-                        message_en="Failed to send test notification.",
-                        message_zh="测试通知发送失败。",
-                    )
-            else:
-                # For other providers, just verify the notifier can be created
+        # Create provider config
+        config = ProviderConfig(
+            type=req.type,
+            enabled=True,
+            token=req.token,
+            chat_id=req.chat_id,
+        )
+        provider = provider_cls(config)
+        async with provider:
+            success, message = await provider.test()
+            if success:
                 return TestResultResponse(
                     success=True,
-                    message_en="Notification configuration is valid.",
-                    message_zh="通知配置有效。",
+                    message_en="Test notification sent successfully.",
+                    message_zh="测试通知发送成功。",
+                )
+            else:
+                return TestResultResponse(
+                    success=False,
+                    message_en=f"Failed to send test notification: {message}",
+                    message_zh=f"测试通知发送失败：{message}",
                 )
     except Exception as e:
         logger.error(f"[Setup] Notification test failed: {e}")
@@ -275,9 +273,14 @@ async def complete_setup(req: SetupCompleteRequest):
         if req.notification_enable:
             config_dict["notification"] = {
                 "enable": True,
-                "type": req.notification_type,
-                "token": req.notification_token,
-                "chat_id": req.notification_chat_id,
+                "providers": [
+                    {
+                        "type": req.notification_type,
+                        "enabled": True,
+                        "token": req.notification_token,
+                        "chat_id": req.notification_chat_id,
+                    }
+                ],
             }
 
         settings.save(config_dict)

backend/src/module/conf/const.py

@@ -36,7 +36,7 @@
         "username": "",
         "password": "",
     },
-    "notification": {"enable": False, "type": "telegram", "token": "", "chat_id": ""},
+    "notification": {"enable": False, "providers": []},
     "experimental_openai": {
         "enable": False,
         "api_key": "",

backend/src/module/core/sub_thread.py

@@ -4,7 +4,7 @@
 from module.conf import settings
 from module.downloader import DownloadClient
 from module.manager import Renamer, TorrentManager, eps_complete
-from module.notification import PostNotification
+from module.notification import NotificationManager
 from module.rss import RSSAnalyser, RSSEngine
 
 from .offset_scanner import OffsetScanner
@@ -66,10 +66,9 @@ async def rename_loop(self):
             async with Renamer() as renamer:
                 renamed_info = await renamer.rename()
             if settings.notification.enable and renamed_info:
-                async with PostNotification() as notifier:
-                    await asyncio.gather(
-                        *[notifier.send_msg(info) for info in renamed_info]
-                    )
+                manager = NotificationManager()
+                for info in renamed_info:
+                    await manager.send_all(info)
             try:
                 await asyncio.wait_for(
                     self.stop_event.wait(),

backend/src/module/models/config.py

@@ -1,7 +1,7 @@
 from os.path import expandvars
-from typing import Literal
+from typing import Literal, Optional
 
-from pydantic import BaseModel, Field, field_validator
+from pydantic import BaseModel, Field, field_validator, model_validator
 
 
 class Program(BaseModel):
@@ -68,19 +68,106 @@ def password(self):
         return expandvars(self.password_)
 
 
+class NotificationProvider(BaseModel):
+    """Configuration for a single notification provider."""
+
+    type: str = Field(..., description="Provider type (telegram, discord, bark, etc.)")
+    enabled: bool = Field(True, description="Whether this provider is enabled")
+
+    # Common fields (with env var expansion)
+    token_: Optional[str] = Field(None, alias="token", description="Auth token")
+    chat_id_: Optional[str] = Field(None, alias="chat_id", description="Chat/channel ID")
+
+    # Provider-specific fields
+    webhook_url_: Optional[str] = Field(
+        None, alias="webhook_url", description="Webhook URL for discord/wecom"
+    )
+    server_url_: Optional[str] = Field(
+        None, alias="server_url", description="Server URL for gotify/bark"
+    )
+    device_key_: Optional[str] = Field(
+        None, alias="device_key", description="Device key for bark"
+    )
+    user_key_: Optional[str] = Field(
+        None, alias="user_key", description="User key for pushover"
+    )
+    api_token_: Optional[str] = Field(
+        None, alias="api_token", description="API token for pushover"
+    )
+    template: Optional[str] = Field(
+        None, description="Custom template for webhook provider"
+    )
+    url_: Optional[str] = Field(
+        None, alias="url", description="URL for generic webhook provider"
+    )
+
+    @property
+    def token(self) -> str:
+        return expandvars(self.token_) if self.token_ else ""
+
+    @property
+    def chat_id(self) -> str:
+        return expandvars(self.chat_id_) if self.chat_id_ else ""
+
+    @property
+    def webhook_url(self) -> str:
+        return expandvars(self.webhook_url_) if self.webhook_url_ else ""
+
+    @property
+    def server_url(self) -> str:
+        return expandvars(self.server_url_) if self.server_url_ else ""
+
+    @property
+    def device_key(self) -> str:
+        return expandvars(self.device_key_) if self.device_key_ else ""
+
+    @property
+    def user_key(self) -> str:
+        return expandvars(self.user_key_) if self.user_key_ else ""
+
+    @property
+    def api_token(self) -> str:
+        return expandvars(self.api_token_) if self.api_token_ else ""
+
+    @property
+    def url(self) -> str:
+        return expandvars(self.url_) if self.url_ else ""
+
+
 class Notification(BaseModel):
-    enable: bool = Field(False, description="Enable notification")
-    type: str = Field("telegram", description="Notification type")
-    token_: str = Field("", alias="token", description="Notification token")
-    chat_id_: str = Field("", alias="chat_id", description="Notification chat id")
+    """Notification configuration supporting multiple providers."""
+
+    enable: bool = Field(False, description="Enable notification system")
+    providers: list[NotificationProvider] = Field(
+        default_factory=list, description="List of notification providers"
+    )
+
+    # Legacy fields for backward compatibility (deprecated)
+    type: Optional[str] = Field(None, description="[Deprecated] Use providers instead")
+    token_: Optional[str] = Field(None, alias="token", description="[Deprecated]")
+    chat_id_: Optional[str] = Field(None, alias="chat_id", description="[Deprecated]")
 
     @property
-    def token(self):
-        return expandvars(self.token_)
+    def token(self) -> str:
+        return expandvars(self.token_) if self.token_ else ""
 
     @property
-    def chat_id(self):
-        return expandvars(self.chat_id_)
+    def chat_id(self) -> str:
+        return expandvars(self.chat_id_) if self.chat_id_ else ""
+
+    @model_validator(mode="after")
+    def migrate_legacy_config(self) -> "Notification":
+        """Auto-migrate old single-provider config to new format."""
+        if self.type and not self.providers:
+            # Old format detected, migrate to new format
+            legacy_provider = NotificationProvider(
+                type=self.type,
+                enabled=True,
+                token=self.token_ or "",
+                chat_id=self.chat_id_ or "",
+            )
+            self.providers = [legacy_provider]
+        return self
 
 
 class ExperimentalOpenAI(BaseModel):

backend/src/module/notification/__init__.py

@@ -1 +1,11 @@
 from .notification import PostNotification
+from .manager import NotificationManager
+from .base import NotificationProvider
+from .providers import PROVIDER_REGISTRY
+
+__all__ = [
+    "PostNotification",
+    "NotificationManager",
+    "NotificationProvider",
+    "PROVIDER_REGISTRY",
+]