Merge pull request #1143 from EstrellaXD/feat/notifier-plugins

feat(plugins): 7 个官方示例插件（WPUSH、OneBot、Apprise、运行外部命令、STRM、Synology Chat、Slack）与 Bark 参数
2026-11-01 13:15:59 +08:00 · 2026-10-21 11:35:46 +02:00
parent f3d42b8f91 f2d93993f4
commit bd3ed917c3
54 changed files with 2166 additions and 34 deletions
--- a/backend/src/module/api/notification.py
+++ b/backend/src/module/api/notification.py
@@ -48,6 +48,10 @@ class TestProviderConfigRequest(BaseModel):
    api_token: Optional[str] = Field(default=None, description="API token")
    template: Optional[str] = Field(default=None, description="Custom template")
    url: Optional[str] = Field(default=None, description="URL for generic webhook")
+    sound: Optional[str] = Field(default=None, description="Bark sound")
+    group: Optional[str] = Field(default=None, description="Bark group")
+    level: Optional[str] = Field(default=None, description="Bark level")
+    icon: Optional[str] = Field(default=None, description="Bark icon URL")


 class TestResponse(BaseModel):
@@ -108,6 +112,10 @@ async def test_provider_config(request: TestProviderConfigRequest):
            api_token=request.api_token or "",
            template=request.template,
            url=request.url or "",
+            sound=request.sound,
+            group=request.group,
+            level=request.level,
+            icon=request.icon,
        )

        success, message = await NotificationManager.test_provider_config(config)
--- a/backend/src/module/models/config.py
+++ b/backend/src/module/models/config.py
@@ -181,7 +181,19 @@ class NotificationProvider(BaseModel):
        ),
    )
    url_: Optional[str] = Field(
-        default=None, alias="url", description="URL for generic webhook provider"
+        default=None,
+        alias="url",
+        description="URL for generic webhook provider; tap-to-open URL for bark",
+    )
+    # Bark 的可选推送参数，留空则不发送
+    sound: Optional[str] = Field(default=None, description="Bark sound name")
+    group: Optional[str] = Field(default=None, description="Bark message group")
+    level: Optional[str] = Field(
+        default=None,
+        description="Bark interruption level: active/timeSensitive/passive/critical",
+    )
+    icon: Optional[str] = Field(
+        default=None, description="Bark icon URL; defaults to the poster URL"
    )

    @property
--- a/backend/src/module/notification/providers/bark.py
+++ b/backend/src/module/notification/providers/bark.py
@@ -22,30 +22,42 @@ class BarkProvider(NotificationProvider):
        self.device_key = config.device_key
        server_url = config.server_url or self.DEFAULT_SERVER
        self.notification_url = f"{server_url.rstrip('/')}/push"
+        # 可选推送参数：只发送已设置的项，旧配置的请求体不变
+        self.options = {
+            "sound": config.sound,
+            "group": config.group,
+            "level": config.level,
+            "url": config.url,
+        }
+        self.icon = config.icon
+
+    def _payload(self, title: str, body: str, icon: str | None = None) -> dict:
+        data: dict[str, str | None] = {
+            "title": title,
+            "body": body,
+            "device_key": self.device_key,
+        }
+        data.update({k: v for k, v in self.options.items() if v})
+        # 配置的图标优先，其次是调用方给的（单集通知的海报）
+        if self.icon or icon:
+            data["icon"] = self.icon or icon
+        return data

    async def send(self, notification: Notification) -> bool:
        """Send notification via Bark."""
        text = self._format_message(notification)
-        data = {
-            "title": notification.official_title,
-            "body": text,
-            "device_key": self.device_key,
-        }
-        poster_url = self._poster_url(notification)
-        if poster_url:
-            data["icon"] = poster_url
-
+        data = self._payload(
+            notification.official_title, text, self._poster_url(notification)
+        )
        resp = await self._post_json(self.notification_url, data)
        logger.debug("Bark notification: %s", resp.status_code)
        return resp.status_code == 200

    async def test(self) -> tuple[bool, str]:
        """Test Bark configuration by sending a test notification."""
-        data = {
-            "title": "AutoBangumi",
-            "body": "通知测试成功！\nNotification test successful!",
-            "device_key": self.device_key,
-        }
+        data = self._payload(
+            "AutoBangumi", "通知测试成功！\nNotification test successful!"
+        )
        try:
            resp = await self._post_json(self.notification_url, data)
            if resp.status_code == 200:
@@ -57,6 +69,5 @@ class BarkProvider(NotificationProvider):

    async def _deliver_text(self, title: str, body: str) -> bool:
        """Deliver a system event via Bark."""
-        data = {"title": title, "body": body, "device_key": self.device_key}
-        resp = await self._post_json(self.notification_url, data)
+        resp = await self._post_json(self.notification_url, self._payload(title, body))
        return resp.status_code == 200
--- a/backend/src/module/notification/resolve.py
+++ b/backend/src/module/notification/resolve.py
@@ -56,7 +56,16 @@ def build_provider(config: "ProviderConfig") -> NotificationProvider | None:
    if entry.plugin_id == CORE:
        # 内置渠道：impl 是 NotificationProvider 子类，直接接收整条配置
        return impl(config)
-    settings = NotifierSettings(
-        template=config.template, extra=dict(config.model_extra or {})
+    # extra = 未知字段 + 已填写的已声明字段（group、icon 等同时是 Bark 的字段，
+    # 声明后不在 model_extra 里）；type/enabled/template 由宿主处理，不放进去。
+    # 已声明字段按别名取同名属性（token、url 等），与内置渠道一样展开 $VAR
+    unknown = config.model_extra or {}
+    declared = config.model_dump(
+        by_alias=True,
+        exclude_none=True,
+        exclude={"type", "enabled", "template", *unknown},
    )
+    expanded = {k: getattr(config, k) for k in declared}
+    extra = {k: v for k, v in expanded.items() if v != ""} | unknown
+    settings = NotifierSettings(template=config.template, extra=extra)
    return PluginNotifier(config, impl(settings))
--- a/backend/src/test/test_notification.py
+++ b/backend/src/test/test_notification.py
@@ -354,6 +354,67 @@ class TestBarkProvider:
        call_args = mock_post.call_args[0]
        assert "device_key" in call_args[1]

+    @pytest.mark.parametrize(
+        ("options", "expected"),
+        [
+            ({}, {}),
+            ({"sound": "", "level": ""}, {}),
+            (
+                {
+                    "sound": "minuet",
+                    "group": "AB",
+                    "level": "timeSensitive",
+                    "url": "https://ab.local",
+                },
+                {
+                    "sound": "minuet",
+                    "group": "AB",
+                    "level": "timeSensitive",
+                    "url": "https://ab.local",
+                },
+            ),
+        ],
+    )
+    async def test_deliver_text_sends_only_set_options(self, options, expected):
+        """#792：sound/group/level/url 只在设置时发送，旧配置的请求体不变。"""
+        config = ProviderConfig(type="bark", device_key="k", **options)
+        provider = BarkProvider(config)
+
+        with patch.object(provider, "_post_json", new_callable=AsyncMock) as mock_post:
+            mock_post.return_value = MagicMock(status_code=200)
+            assert await provider._deliver_text("T", "B") is True
+
+        assert mock_post.call_args[0][1] == {
+            "title": "T",
+            "body": "B",
+            "device_key": "k",
+            **expected,
+        }
+
+    @pytest.mark.parametrize(
+        ("icon", "poster", "expected"),
+        [
+            (None, "https://ab/p.jpg", "https://ab/p.jpg"),
+            ("https://ab/i.png", "https://ab/p.jpg", "https://ab/i.png"),
+            (None, None, None),
+        ],
+    )
+    async def test_send_icon_prefers_configured_icon_over_poster(
+        self, icon, poster, expected
+    ):
+        """配置的 icon 优先；未配置时沿用海报地址；都没有则不发 icon。"""
+        provider = BarkProvider(ProviderConfig(type="bark", device_key="k", icon=icon))
+        notify = Notification(official_title="Test Anime", season=1, episode=5)
+
+        with (
+            patch.object(provider, "_poster_url", return_value=poster),
+            patch.object(provider, "_post_json", new_callable=AsyncMock) as mock_post,
+        ):
+            mock_post.return_value = MagicMock(status_code=200)
+            await provider.send(notify)
+
+        assert mock_post.call_args[0][1].get("icon") == expected
+

 class TestWebhookProvider:
    @pytest.fixture
--- a/backend/src/test/test_plugin_examples.py
+++ b/backend/src/test/test_plugin_examples.py
@@ -117,7 +117,14 @@ async def test_all_examples_load_in_host_with_expected_providers(tmp_path):
    try:
        assert {s.id: s.state for s in mgr.statuses()} == dict.fromkeys(ids, "active")
        assert set(registry.providers(points.RENAME_STRATEGY)) == {"mini-template"}
-        assert set(registry.providers(points.NOTIFIER)) == {"ntfy"}
+        assert set(registry.providers(points.NOTIFIER)) == {
+            "ntfy",
+            "wpush",
+            "onebot",
+            "apprise",
+            "slack",
+            "synology-chat",
+        }
        assert set(registry.providers(points.SEARCH_SITE)) == {"my-tracker"}
        assert [h.plugin_id for h in registry.hooks(points.HTTP_REQUEST)] == [
            "custom-rss-site"
--- a/backend/src/test/test_plugin_extensions.py
+++ b/backend/src/test/test_plugin_extensions.py
@@ -170,6 +170,41 @@ class TestNotifier:
        provider = build_provider(ProviderConfig(type="MyChannel"))
        assert isinstance(provider, PluginNotifier)

+    def test_build_provider_plugin_extra_includes_set_declared_fields(self, registry):
+        from module.notification.resolve import build_provider
+
+        created: list[RecordingNotifier] = []
+
+        def factory(conf: NotifierSettings) -> RecordingNotifier:
+            created.append(RecordingNotifier(conf))
+            return created[-1]
+
+        add_plugin_provider(registry, points.NOTIFIER, "chat", factory)
+        # group/icon 是 Bark 声明的字段，不在 model_extra 里；未填写的 sound 不传
+        cfg = ProviderConfig.model_validate(
+            {"type": "chat", "room": "42", "group": "g", "icon": "i", "sound": ""}
+        )
+        build_provider(cfg)
+        assert created[0].conf.extra == {"room": "42", "group": "g", "icon": "i"}
+
+    def test_build_provider_plugin_extra_expands_env_vars_in_declared_fields(
+        self, registry, monkeypatch
+    ):
+        from module.notification.resolve import build_provider
+
+        created: list[RecordingNotifier] = []
+
+        def factory(conf: NotifierSettings) -> RecordingNotifier:
+            created.append(RecordingNotifier(conf))
+            return created[-1]
+
+        add_plugin_provider(registry, points.NOTIFIER, "chat", factory)
+        monkeypatch.setenv("BOT_TOKEN", "abc")
+        # 内置渠道读 config.token 时展开 $VAR，插件拿到的 extra 也应一致
+        cfg = ProviderConfig.model_validate({"type": "chat", "token": "$BOT_TOKEN"})
+        build_provider(cfg)
+        assert created[0].conf.extra == {"token": "abc"}
+
    def test_extra_fields_survive_config_roundtrip(self):
        cfg = ProviderConfig.model_validate({"type": "chat", "room": "42"})
        assert cfg.model_dump(by_alias=True)["room"] == "42"
--- a/docs/config/notifier.md
+++ b/docs/config/notifier.md
@@ -24,7 +24,7 @@
 - Telegram：`Bot Token`、`Chat ID`
 - Discord：Webhook URL
 - WeCom：Webhook URL、Key
- Bark：Device Key，可选 Server URL
+- Bark：Device Key，可选 Server URL、铃声、分组、中断级别、图标地址、点击打开的地址
 - Server Chan：SendKey
 - Gotify：Server URL、App Token
 - Pushover：User Key、API Token
@@ -53,9 +53,10 @@ Webhook 模板可以使用 `{{title}}`、`{{season}}`、`{{episode}}`、`{{poste
 | `token` | Telegram（Bot Token）、Server Chan（SendKey）、WeCom（Key）、Gotify（App Token）使用 |
 | `chat_id` | Telegram 使用 |
 | `webhook_url` | Discord、WeCom 使用 |
-| `url` | Webhook 使用 |
+| `url` | Webhook 使用；Bark 中为点击通知后打开的地址（可选） |
 | `server_url` | Bark（可选）、Gotify 使用 |
 | `device_key` | Bark 使用 |
+| `sound` / `group` / `level` / `icon` | Bark 可选：铃声、分组、中断级别（`active`、`timeSensitive`、`passive`、`critical`）、图标地址。留空则不发送；`icon` 留空时单集通知用海报地址 |
 | `user_key` / `api_token` | Pushover 使用 |
 | `template` | 自定义通知模板 |

--- a/docs/dev/plugins/examples.md
+++ b/docs/dev/plugins/examples.md
@@ -9,6 +9,13 @@
 | `template-rename` | `rename_strategy` Provider，自带过滤器的文件名模板；`RenameSkipped`；`RenameStrategyContract` |
 | `nfo-writer` | 订阅 `torrent.organized`，在正片旁写 `.nfo`；路径映射；幂等 |
 | `ntfy-notifier` | `notifier` Provider；后端拒绝时返回 `False`；`NotifierContract` |
+| `wpush-notifier` | `notifier` Provider；WPUSH 多渠道推送，按返回 JSON 的 `code` 判断成败 |
+| `onebot-notifier` | `notifier` Provider；OneBot v11 私聊 / 群消息，消息段附带海报 `poster_url` |
+| `apprise-notifier` | `notifier` Provider；调用 Apprise API 服务器，支持 URL 列表与配置键两种方式 |
+| `run-command` | 订阅 `torrent.organized` / `file.renamed`，运行外部命令；参数列表不经过 shell；环境变量与占位符；超时终止 |
+| `strm-writer` | 订阅 `torrent.organized`，按媒体目录结构写 `.strm`；幂等 |
+| `synology-chat-notifier` | `notifier` Provider；表单字段 `payload=<JSON>`；HTTP 200 但 `success: false` 时返回 `False` |
+| `slack-notifier` | `notifier` Provider；mrkdwn 转义；带海报时附图片块 |
 | `manual-pick` | 前端挂载点 `bangumi.detail.tab`、插件路由、插件 KV、事件 |

 ## 安装一个示例
--- a/docs/en/config/notifier.md
+++ b/docs/en/config/notifier.md
@@ -24,7 +24,7 @@ Provider-specific fields include:
 - Telegram: `Bot Token`, `Chat ID`
 - Discord: Webhook URL
 - WeCom: Webhook URL and Key
- Bark: Device Key and optional Server URL
+- Bark: Device Key. Optional: Server URL, sound, group, interruption level, icon URL, and the URL to open on tap
 - Server Chan: SendKey
 - Gotify: Server URL and App Token
 - Pushover: User Key and API Token
@@ -53,9 +53,10 @@ Each object in `providers` usually contains:
 | `token` | Used by Telegram (Bot Token), Server Chan (SendKey), WeCom (Key) and Gotify (App Token) |
 | `chat_id` | Used by Telegram |
 | `webhook_url` | Used by Discord and WeCom |
-| `url` | Used by Webhook |
+| `url` | Used by Webhook. For Bark, the URL to open on tap (optional) |
 | `server_url` | Used by Bark (optional) and Gotify |
 | `device_key` | Used by Bark |
+| `sound` / `group` / `level` / `icon` | Optional for Bark: sound, group, interruption level (`active`, `timeSensitive`, `passive`, `critical`), and icon URL. AB does not send an empty field. When `icon` is empty, episode notifications use the poster URL |
 | `user_key` / `api_token` | Used by Pushover |
 | `template` | Custom message template |

--- a/docs/en/dev/plugins/examples.md
+++ b/docs/en/dev/plugins/examples.md
@@ -9,6 +9,10 @@ The examples are in [`examples/plugins/`](https://github.com/EstrellaXD/Auto_Ban
 | `template-rename` | A `rename_strategy` Provider with a file-name template that has its own filters. `RenameSkipped`. `RenameStrategyContract` |
 | `nfo-writer` | Subscribes to `torrent.organized` and writes an `.nfo` next to each media file. Path mapping. Idempotent |
 | `ntfy-notifier` | A `notifier` Provider. Returns `False` when the backend refuses. `NotifierContract` |
+| `run-command` | Subscribes to `torrent.organized` / `file.renamed` and runs an external command. The argument list does not go through a shell. Environment variables and placeholders. Kill on timeout |
+| `strm-writer` | Subscribes to `torrent.organized` and writes `.strm` files that mirror the media folders. Idempotent |
+| `synology-chat-notifier` | A `notifier` Provider. Sends the form field `payload=<JSON>`. Returns `False` on HTTP 200 with `success: false` |
+| `slack-notifier` | A `notifier` Provider. Escapes mrkdwn. Adds an image block when a poster is available |
 | `manual-pick` | The frontend slot `bangumi.detail.tab`, plugin routes, plugin KV, events |

 ## Install an example
--- a/docs/ja/config/notifier.md
+++ b/docs/ja/config/notifier.md
@@ -42,9 +42,10 @@ Webhookテンプレートでは `{{title}}`、`{{season}}`、`{{episode}}`、`{{
 | `token` | Telegram（Bot Token）、Server Chan（SendKey）、WeCom（Key）、Gotify（App Token）で使用 |
 | `chat_id` | Telegram で使用 |
 | `webhook_url` | Discord、WeCom で使用 |
-| `url` | Webhook で使用 |
+| `url` | Webhook で使用。Bark ではタップで開く URL（任意） |
 | `server_url` | Bark（任意）、Gotify で使用 |
 | `device_key` | Bark で使用 |
+| `sound` / `group` / `level` / `icon` | Bark（任意）：通知音、グループ、割り込みレベル（`active`、`timeSensitive`、`passive`、`critical`）、アイコン URL。空の項目は送信しません。`icon` が空のとき、エピソード通知はポスターの URL を使います |
 | `user_key` / `api_token` | Pushover で使用 |
 | `template` | カスタム通知テンプレート |

--- a/docs/ja/dev/plugins/examples.md
+++ b/docs/ja/dev/plugins/examples.md
@@ -9,6 +9,10 @@
 | `template-rename` | `rename_strategy` Provider。独自フィルター付きのファイル名テンプレート。`RenameSkipped`。`RenameStrategyContract` |
 | `nfo-writer` | `torrent.organized` を購読し、本編の隣に `.nfo` を書きます。パス変換。冪等 |
 | `ntfy-notifier` | `notifier` Provider。バックエンドが拒否したときは `False` を返します。`NotifierContract` |
+| `run-command` | `torrent.organized` / `file.renamed` を購読し、外部コマンドを実行します。引数リストはシェルを通しません。環境変数とプレースホルダー。タイムアウトで終了 |
+| `strm-writer` | `torrent.organized` を購読し、メディアのフォルダー構成どおりに `.strm` を書きます。冪等 |
+| `synology-chat-notifier` | `notifier` Provider。フォームフィールド `payload=<JSON>` を送ります。HTTP 200 でも `success: false` なら `False` を返します |
+| `slack-notifier` | `notifier` Provider。mrkdwn をエスケープします。ポスターがあるときは画像ブロックを付けます |
 | `manual-pick` | フロントエンドスロット `bangumi.detail.tab`、プラグインのルート、プラグイン KV、イベント |

 ## サンプルをインストールする
--- a/examples/plugins/apprise-notifier/README.md
+++ b/examples/plugins/apprise-notifier/README.md
@@ -0,0 +1,45 @@
+# Apprise 通知
+
+通过 [Apprise API](https://github.com/caronc/apprise-api) 服务器把通知转发到 Telegram、Discord、邮件等上百种服务。插件不依赖 `apprise` 库，只调用 HTTP 接口。
+
+- `@provider(points.NOTIFIER, id="apprise")`：返回 `NotifierFactory`
+- 填写了 Apprise URL：`POST {server}/notify/`，请求体 `{"urls", "title", "body"}`，多个 URL 以逗号拼接成一个字符串
+- 否则使用配置键：`POST {server}/notify/{config_key}`，请求体 `{"title", "body"}`，URL 保存在服务器上
+- 两者都没填时 `send` 返回 `False`，「测试」提示需要填写其中之一
+- HTTP 2xx 即成功
+
+## 准备 Apprise API 服务器
+
+自行运行 `caronc/apprise` 容器，例如：
+
+```bash
+docker run -d --name apprise -p 8000:8000 caronc/apprise:latest
+```
+
+与 AutoBangumi 在同一 Docker 网络时，服务器地址可填 `http://apprise:8000`。使用配置键时，先在 Apprise 的网页界面（`http://<服务器>:8000/cfg/<配置键>`）保存通知 URL。
+
+## 配置
+
+| 字段 | 说明 |
+| --- | --- |
+| `server` | Apprise API 服务器地址，默认 `http://apprise:8000` |
+| `urls` | 可选：Apprise URL，如 `tgram://bottoken/ChatID`；多个 URL 用逗号或换行分隔；填写后忽略配置键。URL 含令牌，按秘密字段处理，读取配置时只返回掩码 |
+| `config_key` | 可选：服务器上保存的配置键 |
+
+## 安装
+
+```bash
+cp -r examples/plugins/apprise-notifier config/plugins/local/
+```
+
+在「设置 → 插件」中开启「允许未签名插件」，启用并填写服务器与 URL（或配置键）；再到「设置 → 通知 → 添加渠道」选择类型 `apprise`。
+
+## 测试
+
+```bash
+cd examples/plugins/apprise-notifier && uv run pytest
+```
+
+---
+
+基于 @LTSlw 在 [#942](https://github.com/EstrellaXD/Auto_Bangumi/pull/942) 中的实现
--- a/examples/plugins/apprise-notifier/apprise_notifier/init.py
+++ b/examples/plugins/apprise-notifier/apprise_notifier/init.py
@@ -0,0 +1,93 @@
+"""Apprise 通知渠道。
+
+不依赖 apprise 库，而是调用用户自建的 Apprise API 服务器（caronc/apprise 镜像）：
+
+- 填了 ``urls``：无状态调用 ``POST {server}/notify/``，请求体带上 Apprise URL；
+- 否则用 ``config_key``：``POST {server}/notify/{config_key}``，URL 存在服务器上。
+
+HTTP 2xx 即成功。接口文档：https://github.com/caronc/apprise-api
+"""
+
+from collections.abc import Callable
+from typing import Any
+
+import httpx2
+from pydantic import BaseModel, Field
+
+from ab_sdk import Plugin, points, provider, secret_field
+from ab_sdk.notify import NotificationMessage, NotifierFactory, NotifierSettings
+
+
+class Options(BaseModel):
+    server: str = Field(
+        "http://apprise:8000", title="服务器", description="Apprise API 服务器地址"
+    )
+    # URL 里带着令牌，按秘密字段掩码
+    urls: str = secret_field(
+        title="Apprise URL",
+        description="多个 Apprise URL 用逗号或换行分隔",
+    )
+    config_key: str = Field(
+        "",
+        title="配置键",
+        description="未填 Apprise URL 时，使用服务器上保存的这份配置",
+    )
+
+
+class Channel:
+    def __init__(
+        self,
+        options: Options,
+        settings: NotifierSettings,
+        client_factory: Callable[[], httpx2.AsyncClient],
+    ) -> None:
+        self.options = options
+        self.settings = settings
+        self.client_factory = client_factory
+
+    async def _post(self, message: NotificationMessage) -> tuple[bool, str]:
+        opts = self.options
+        server = opts.server.rstrip("/")
+        payload: dict[str, Any] = {"title": message.title, "body": message.body}
+        urls = [u.strip() for u in opts.urls.replace("\n", ",").split(",")]
+        urls = [u for u in urls if u]
+        if urls:
+            url = f"{server}/notify/"
+            # Apprise 接受逗号分隔的 URL 字符串
+            payload["urls"] = ",".join(urls)
+        elif opts.config_key:
+            url = f"{server}/notify/{opts.config_key}"
+        else:
+            return False, "需填写 Apprise URL 或配置键"
+        try:
+            async with self.client_factory() as client:
+                resp = await client.post(url, json=payload)
+        except httpx2.HTTPError as e:
+            return False, f"请求失败：{e}"
+        if resp.is_success:
+            return True, "ok"
+        return False, f"Apprise 返回 HTTP {resp.status_code}：{resp.text[:200]}"
+
+    async def send(self, message: NotificationMessage) -> bool:
+        ok, _ = await self._post(message)
+        return ok
+
+    async def test(self) -> tuple[bool, str]:
+        return await self._post(
+            NotificationMessage(
+                kind="event", title="AutoBangumi 通知测试", body="通知测试成功！"
+            )
+        )
+
+
+class AppriseNotifier(Plugin[Options]):
+    config_model = Options
+
+    # 测试可替换为带 MockTransport 的客户端
+    client_factory: Callable[[], httpx2.AsyncClient] = staticmethod(
+        lambda: httpx2.AsyncClient(timeout=10)
+    )
+
+    @provider(points.NOTIFIER, id="apprise")
+    def channel(self) -> NotifierFactory:
+        return lambda settings: Channel(self.config, settings, self.client_factory)
--- a/examples/plugins/apprise-notifier/plugin.toml
+++ b/examples/plugins/apprise-notifier/plugin.toml
@@ -0,0 +1,10 @@
+[plugin]
+id = "apprise-notifier"
+name = "Apprise 通知"
+version = "0.1.0"
+sdk = ">=0.5,<1"
+entry = "apprise_notifier:AppriseNotifier"
+description = "通过 Apprise API 服务器（caronc/apprise）转发通知到上百种服务。在「设置 → 通知 → 添加渠道」选择类型 apprise"
+authors = ["AutoBangumi"]
+permissions = ["network"]
+extension_points = ["notifier"]
--- a/examples/plugins/apprise-notifier/pyproject.toml
+++ b/examples/plugins/apprise-notifier/pyproject.toml
@@ -0,0 +1,13 @@
+# 只用于本地开发与测试，不会被打进插件包（ab-plugin pack 排除它）
+[project]
+name = "ab-plugin-apprise-notifier"
+version = "0.1.0"
+requires-python = ">=3.14"
+dependencies = ["autobangumi-sdk"]
+
+[dependency-groups]
+dev = ["pytest>=8.0.0"]
+
+[tool.pytest.ini_options]
+pythonpath = ["."]
+testpaths = ["tests"]
--- a/examples/plugins/apprise-notifier/tests/test_apprise_notifier.py
+++ b/examples/plugins/apprise-notifier/tests/test_apprise_notifier.py
@@ -0,0 +1,70 @@
+import asyncio
+import json
+
+import httpx2
+import pytest
+from apprise_notifier import AppriseNotifier, Options
+
+from ab_sdk.notify import NotificationMessage, NotifierSettings
+from ab_sdk.testing import NotifierContract, create_plugin
+
+MESSAGE = NotificationMessage(kind="event", title="标题", body="正文")
+
+
+def channel(status: int, options: dict, seen=None):
+    def handler(request: httpx2.Request) -> httpx2.Response:
+        if seen is not None:
+            seen.append(request)
+        return httpx2.Response(status)
+
+    plugin, _ = create_plugin(AppriseNotifier, options)
+    plugin.client_factory = lambda: httpx2.AsyncClient(  # type: ignore[method-assign]
+        transport=httpx2.MockTransport(handler)
+    )
+    return plugin.channel()(NotifierSettings())
+
+
+class TestChannel(NotifierContract):
+    def create(self):
+        return channel(200, {"config_key": "ab"})
+
+    def create_failing(self):
+        return channel(424, {"config_key": "ab"})
+
+
+@pytest.mark.parametrize(
+    ("options", "path", "payload"),
+    [
+        (
+            {"server": "http://ap:8000/", "urls": "tgram://t/1,\n mailto://u ,"},
+            "/notify/",
+            {"title": "标题", "body": "正文", "urls": "tgram://t/1,mailto://u"},
+        ),
+        (
+            {"server": "http://ap:8000", "config_key": "ab"},
+            "/notify/ab",
+            {"title": "标题", "body": "正文"},
+        ),
+    ],
+)
+def test_send_posts_stateless_urls_or_config_key(options, path, payload):
+    seen: list[httpx2.Request] = []
+    assert asyncio.run(channel(200, options, seen).send(MESSAGE)) is True
+    [request] = seen
+    assert request.url.path == path
+    assert json.loads(request.content) == payload
+
+
+def test_missing_urls_and_config_key_fails_without_request():
+    seen: list[httpx2.Request] = []
+    ch = channel(200, {}, seen)
+    assert asyncio.run(ch.send(MESSAGE)) is False
+    assert asyncio.run(ch.test()) == (False, "需填写 Apprise URL 或配置键")
+    assert seen == []
+
+
+def test_urls_is_secret_in_config_schema():
+    # URL 里带着令牌：宿主的读配置接口按 schema 中的 secret 标记掩码
+    urls = Options.model_json_schema()["properties"]["urls"]
+    assert urls["secret"] is True
+    assert urls["type"] == "string"
--- a/examples/plugins/onebot-notifier/README.md
+++ b/examples/plugins/onebot-notifier/README.md
@@ -0,0 +1,40 @@
+# OneBot 通知
+
+通过 [OneBot v11](https://github.com/botuniverse/onebot-11) HTTP API 把通知发到 QQ 私聊或群。NapCat、Lagrange.OneBot、LLOneBot 等实现都提供这个接口。
+
+- `@provider(points.NOTIFIER, id="onebot")`：返回 `NotifierFactory`
+- 向 `{url}/send_private_msg`（`user_id`）或 `{url}/send_group_msg`（`group_id`）POST JSON，消息为消息段数组
+- 填写了 Access Token 时附带 `Authorization: Bearer <token>`
+- HTTP 2xx 且返回 `status == "ok"` 或 `retcode == 0` 才算成功
+
+## 配置
+
+| 字段 | 说明 |
+| --- | --- |
+| `url` | OneBot HTTP API 地址，如 `http://napcat:3000` |
+| `access_token` | 可选：Access Token（秘密字段） |
+| `message_type` | `private`（私聊）或 `group`（群聊） |
+| `target` | 私聊填 QQ 号，群聊填群号 |
+| `send_poster` | 是否附带海报图片，默认开启 |
+
+## 海报
+
+插件只能拿到宿主给出的公开海报地址，拿不到本地海报文件。因此要发送海报，需要在「设置 → 通知」中配置 `base_url`，并保证 OneBot 实现能访问这个地址；它会自行下载图片。未配置时只发送文字。海报在文字之后作为单独一条消息发送：发送结果只看文字，图片失败（如 OneBot 实现取不到海报）不影响已送达的文字，也不重发。
+
+## 安装
+
+```bash
+cp -r examples/plugins/onebot-notifier config/plugins/local/
+```
+
+在「设置 → 插件」中开启「允许未签名插件」，启用并填写 API 地址与目标；再到「设置 → 通知 → 添加渠道」选择类型 `onebot`。
+
+## 测试
+
+```bash
+cd examples/plugins/onebot-notifier && uv run pytest
+```
+
+---
+
+基于 @AMYdd00 在 [#1033](https://github.com/EstrellaXD/Auto_Bangumi/pull/1033) 中的实现
--- a/examples/plugins/onebot-notifier/onebot_notifier/init.py
+++ b/examples/plugins/onebot-notifier/onebot_notifier/init.py
@@ -0,0 +1,128 @@
+"""OneBot v11 通知渠道。
+
+POST JSON 到 ``{url}/send_private_msg`` 或 ``{url}/send_group_msg``，消息为消息段
+数组。先发文字；文字成功且开启海报、宿主给出海报地址时，再单独发一条图片消息。
+HTTP 2xx 且返回 ``status == "ok"`` 或 ``retcode == 0`` 才算成功，发送结果只看文字：
+OneBot 实现取不到海报时只丢图片，文字已经送达，不重发。
+协议文档：https://github.com/botuniverse/onebot-11
+"""
+
+from collections.abc import Callable
+from contextlib import suppress
+from typing import Any, Literal
+
+import httpx2
+from pydantic import BaseModel, Field
+
+from ab_sdk import Plugin, points, provider, secret_field
+from ab_sdk.notify import NotificationMessage, NotifierFactory, NotifierSettings
+
+
+class Options(BaseModel):
+    url: str = Field(
+        "",
+        title="API 地址",
+        description="OneBot v11 HTTP API 地址，如 http://napcat:3000",
+    )
+    access_token: str = secret_field(
+        title="Access Token", description="可选：以 Authorization: Bearer 发送"
+    )
+    message_type: Literal["private", "group"] = Field("private", title="消息类型")
+    target: int = Field(0, title="目标", description="私聊填 QQ 号，群聊填群号")
+    send_poster: bool = Field(
+        True,
+        title="发送海报",
+        description="需在通知设置里配置 base_url，宿主才会给出可访问的海报地址",
+    )
+
+
+class Channel:
+    def __init__(
+        self,
+        options: Options,
+        settings: NotifierSettings,
+        client_factory: Callable[[], httpx2.AsyncClient],
+    ) -> None:
+        self.options = options
+        self.settings = settings
+        self.client_factory = client_factory
+
+    async def _post(self, message: NotificationMessage) -> tuple[bool, str]:
+        opts = self.options
+        if not opts.url or not opts.target:
+            return False, "未填写 API 地址或目标"
+        # 单集通知的正文已含番剧名；系统事件的标题与正文分开，拼在一起发
+        text = (
+            message.body
+            if message.kind == "episode"
+            else f"{message.title}\n{message.body}"
+        )
+        try:
+            async with self.client_factory() as client:
+                ok, detail = await self._send(
+                    client, [{"type": "text", "data": {"text": text}}]
+                )
+                # SDK 只给出公开海报 URL（没有本地路径），由 OneBot 实现自行下载；
+                # 图片失败（含请求异常）不影响已送达的文字
+                if ok and opts.send_poster and message.poster_url:
+                    image = {"type": "image", "data": {"file": message.poster_url}}
+                    with suppress(httpx2.HTTPError):
+                        await self._send(client, [image])
+        except httpx2.HTTPError as e:
+            return False, f"请求失败：{e}"
+        return ok, detail
+
+    async def _send(
+        self, client: httpx2.AsyncClient, segments: list[dict[str, Any]]
+    ) -> tuple[bool, str]:
+        opts = self.options
+        id_key = "group_id" if opts.message_type == "group" else "user_id"
+        headers = {}
+        if opts.access_token:
+            headers["Authorization"] = f"Bearer {opts.access_token}"
+        resp = await client.post(
+            f"{opts.url.rstrip('/')}/send_{opts.message_type}_msg",
+            json={id_key: opts.target, "message": segments},
+            headers=headers,
+        )
+        try:
+            body = resp.json()
+        except ValueError:
+            body = None
+        if (
+            resp.is_success
+            and isinstance(body, dict)
+            and (body.get("status") == "ok" or body.get("retcode") == 0)
+        ):
+            return True, "ok"
+        detail = (
+            (body.get("wording") or body.get("msg")) if isinstance(body, dict) else None
+        )
+        if not resp.is_success:
+            # 反向代理的错误页不是 JSON，先报 HTTP 状态
+            detail = f"HTTP {resp.status_code}" + (f"，{detail}" if detail else "")
+        return False, f"OneBot 返回错误：{detail or '响应不是预期的 JSON'}"
+
+    async def send(self, message: NotificationMessage) -> bool:
+        ok, _ = await self._post(message)
+        return ok
+
+    async def test(self) -> tuple[bool, str]:
+        return await self._post(
+            NotificationMessage(
+                kind="event", title="AutoBangumi 通知测试", body="通知测试成功！"
+            )
+        )
+
+
+class OneBotNotifier(Plugin[Options]):
+    config_model = Options
+
+    # 测试可替换为带 MockTransport 的客户端
+    client_factory: Callable[[], httpx2.AsyncClient] = staticmethod(
+        lambda: httpx2.AsyncClient(timeout=10)
+    )
+
+    @provider(points.NOTIFIER, id="onebot")
+    def channel(self) -> NotifierFactory:
+        return lambda settings: Channel(self.config, settings, self.client_factory)
--- a/examples/plugins/onebot-notifier/plugin.toml
+++ b/examples/plugins/onebot-notifier/plugin.toml
@@ -0,0 +1,10 @@
+[plugin]
+id = "onebot-notifier"
+name = "OneBot 通知"
+version = "0.1.0"
+sdk = ">=0.5,<1"
+entry = "onebot_notifier:OneBotNotifier"
+description = "通过 OneBot v11 HTTP API（NapCat、Lagrange 等）把通知发到 QQ 私聊或群。在「设置 → 通知 → 添加渠道」选择类型 onebot"
+authors = ["AutoBangumi"]
+permissions = ["network"]
+extension_points = ["notifier"]
--- a/examples/plugins/onebot-notifier/pyproject.toml
+++ b/examples/plugins/onebot-notifier/pyproject.toml
@@ -0,0 +1,13 @@
+# 只用于本地开发与测试，不会被打进插件包（ab-plugin pack 排除它）
+[project]
+name = "ab-plugin-onebot-notifier"
+version = "0.1.0"
+requires-python = ">=3.14"
+dependencies = ["autobangumi-sdk"]
+
+[dependency-groups]
+dev = ["pytest>=8.0.0"]
+
+[tool.pytest.ini_options]
+pythonpath = ["."]
+testpaths = ["tests"]
--- a/examples/plugins/onebot-notifier/tests/test_onebot_notifier.py
+++ b/examples/plugins/onebot-notifier/tests/test_onebot_notifier.py
@@ -0,0 +1,148 @@
+import asyncio
+import json
+from collections.abc import Callable
+
+import httpx2
+import pytest
+from onebot_notifier import OneBotNotifier
+
+from ab_sdk.notify import NotificationMessage, NotifierSettings
+from ab_sdk.testing import NotifierContract, create_plugin
+
+BASE = {"url": "http://bot:3000/", "target": 123}
+POSTER = "https://ab.example.com/posters/a.jpg"
+
+
+def channel(
+    status: int,
+    body: dict | str | Callable[[dict], httpx2.Response],
+    options: dict | None = None,
+    seen=None,
+):
+    def handler(request: httpx2.Request) -> httpx2.Response:
+        if seen is not None:
+            seen.append(request)
+        if callable(body):
+            return body(json.loads(request.content))
+        if isinstance(body, str):
+            return httpx2.Response(status, text=body)
+        return httpx2.Response(status, json=body)
+
+    plugin, _ = create_plugin(OneBotNotifier, BASE if options is None else options)
+    plugin.client_factory = lambda: httpx2.AsyncClient(  # type: ignore[method-assign]
+        transport=httpx2.MockTransport(handler)
+    )
+    return plugin.channel()(NotifierSettings())
+
+
+def episode(poster_url: str | None = None) -> NotificationMessage:
+    return NotificationMessage(
+        kind="episode", title="番剧", body="番剧 第1集", poster_url=poster_url
+    )
+
+
+class TestChannel(NotifierContract):
+    def create(self):
+        return channel(200, {"status": "ok", "retcode": 0})
+
+    def create_failing(self):
+        return channel(200, {"status": "failed", "retcode": 100, "wording": "不是好友"})
+
+
+@pytest.mark.parametrize(
+    ("message_type", "path", "id_key"),
+    [
+        ("private", "/send_private_msg", "user_id"),
+        ("group", "/send_group_msg", "group_id"),
+    ],
+)
+def test_send_posts_to_endpoint_with_target(message_type, path, id_key):
+    seen: list[httpx2.Request] = []
+    ch = channel(200, {"retcode": 0}, {**BASE, "message_type": message_type}, seen)
+    assert asyncio.run(ch.send(episode())) is True
+    [request] = seen
+    assert request.url.path == path
+    assert json.loads(request.content) == {
+        id_key: 123,
+        "message": [{"type": "text", "data": {"text": "番剧 第1集"}}],
+    }
+
+
+TEXT = [{"type": "text", "data": {"text": "番剧 第1集"}}]
+IMAGE = [{"type": "image", "data": {"file": POSTER}}]
+
+
+@pytest.mark.parametrize(
+    ("send_poster", "poster_url", "messages"),
+    [
+        (True, POSTER, [TEXT, IMAGE]),
+        (False, POSTER, [TEXT]),
+        (True, None, [TEXT]),
+    ],
+)
+def test_send_poster_goes_as_separate_message_after_text(
+    send_poster, poster_url, messages
+):
+    seen: list[httpx2.Request] = []
+    ch = channel(200, {"status": "ok"}, {**BASE, "send_poster": send_poster}, seen)
+    assert asyncio.run(ch.send(episode(poster_url))) is True
+    assert [json.loads(r.content)["message"] for r in seen] == messages
+
+
+@pytest.mark.parametrize(("token", "header"), [("t0k", "Bearer t0k"), ("", None)])
+def test_send_sets_bearer_header_only_with_token(token, header):
+    seen: list[httpx2.Request] = []
+    ch = channel(200, {"status": "ok"}, {**BASE, "access_token": token}, seen)
+    assert asyncio.run(ch.send(episode())) is True
+    assert seen[0].headers.get("Authorization") == header
+
+
+def test_test_reports_failure_on_http_error():
+    ok, detail = asyncio.run(channel(401, {"msg": "token 错误"}).test())
+    assert ok is False
+    assert "token 错误" in detail
+
+
+def test_send_without_target_returns_false_without_request():
+    seen: list[httpx2.Request] = []
+    ch = channel(200, {"status": "ok"}, {"url": "http://bot:3000"}, seen)
+    assert asyncio.run(ch.send(episode())) is False
+    assert seen == []
+
+
+# OneBot 实现取不到海报：图片消息返回失败，或请求本身出错
+def image_rejected(body: dict) -> httpx2.Response:
+    if body["message"] == IMAGE:
+        return httpx2.Response(200, json={"status": "failed", "retcode": 1200})
+    return httpx2.Response(200, json={"status": "ok", "retcode": 0})
+
+
+def image_unreachable(body: dict) -> httpx2.Response:
+    if body["message"] == IMAGE:
+        raise httpx2.ReadTimeout("timeout")
+    return httpx2.Response(200, json={"status": "ok", "retcode": 0})
+
+
+@pytest.mark.parametrize("respond", [image_rejected, image_unreachable])
+def test_send_image_failure_keeps_text_result_without_retry(respond):
+    seen: list[httpx2.Request] = []
+    ch = channel(200, respond, seen=seen)
+    assert asyncio.run(ch.send(episode(POSTER))) is True
+    assert [json.loads(r.content)["message"] for r in seen] == [TEXT, IMAGE]
+
+
+def test_send_text_failure_skips_image_and_is_not_retried():
+    seen: list[httpx2.Request] = []
+    ch = channel(200, {"status": "failed", "retcode": 100}, seen=seen)
+    assert asyncio.run(ch.send(episode(POSTER))) is False
+    assert len(seen) == 1
+
+
+@pytest.mark.parametrize(
+    ("status", "body", "expected"),
+    [(502, "<html>Bad Gateway</html>", "HTTP 502"), (200, "<html>", "不是预期的 JSON")],
+)
+def test_test_non_json_response_reports_status(status, body, expected):
+    ok, detail = asyncio.run(channel(status, body).test())
+    assert ok is False
+    assert expected in detail
--- a/examples/plugins/run-command/README.md
+++ b/examples/plugins/run-command/README.md
@@ -0,0 +1,57 @@
+# 运行外部命令（示例插件）
+
+种子整理完成（`torrent.organized`）或文件被重命名（`file.renamed`）后运行一条外部命令，例如自己的清理脚本、FileBot、hlink。演示：
+
+- `@subscribe("*", timeout=...)` 订阅全部事件，按配置过滤；处理超时放宽到命令超时之上
+- `asyncio.create_subprocess_exec` 直接执行参数列表，不经过 shell
+- 命令超时被杀掉，退出码非 0 时记录 stderr 的最后 2000 字节（边读边丢，输出再多也不占满内存）；命令退出后插件即关闭 stderr，留在后台的进程需自行重定向输出。失败只记日志，不让插件被熔断
+
+## 配置
+
+| 字段 | 说明 |
+| --- | --- |
+| 命令 | 程序及其参数，每项一个，如 `/config/scripts/link.sh`、`{files}`。不经过 shell，管道、重定向、`$变量` 都不会展开；需要时把命令写成脚本 |
+| 事件 | 在哪些事件后运行，默认只有 `torrent.organized` |
+| 超时（秒） | 默认 300，最大 3600。超时后杀掉命令 |
+
+## 传给命令的字段
+
+每个字段都是环境变量。参数列表里某一项**整项**写成占位符（环境变量名去掉 `AB_` 后的小写，如 `{torrent_hash}`）时，这一项替换为字段值；`{files}` 展开为每个文件一个参数。只替换整项：`--hash={torrent_hash}`、`sh -c "echo {new_path}"` 这类夹在参数里的占位符不替换，原样传给程序；未知的占位符也原样保留。
+
+文件名来自 RSS，可能含有 `;`、`$()`、引号。不要把占位符写进 `sh -c` 的脚本文本，脚本里从 `AB_*` 环境变量读取事件数据，并给变量加双引号：
+
+```sh
+#!/bin/sh
+# /config/scripts/on-rename.sh，命令填 ["/config/scripts/on-rename.sh"]
+ln -f "$AB_NEW_PATH" "/media/links/$(basename "$AB_NEW_PATH")"
+```
+
+| 环境变量 | 事件 | 内容 |
+| --- | --- | --- |
+| `AB_EVENT` | 全部 | 事件名 |
+| `AB_BANGUMI_ID` | 全部 | 番剧 id；旧种子没有 `ab:<id>` 标签时为空 |
+| `AB_DOWNLOADER_ID` | 全部 | 下载器实例 id |
+| `AB_TORRENT_HASH` | `torrent.organized` | 种子 hash |
+| `AB_FILES` | `torrent.organized` | 正片与字幕的最终路径，每行一个。路径本身含换行时无法按行区分，需要准确的路径请用 `{files}` 参数 |
+| `AB_OLD_PATH` / `AB_NEW_PATH` | `file.renamed` | 重命名前后的路径 |
+| `AB_FILE_KIND` | `file.renamed` | `media` 或 `subtitle` |
+
+命令继承 AutoBangumi 的环境变量，但 AutoBangumi 自己的 `AB_*`（如下载器密码）不传给命令。事件里没有种子的保存目录，需要时从 `AB_FILES` 取。路径是**下载器视角**，AutoBangumi 与下载器的挂载不同时，脚本要自己换算。命令按事件顺序逐个运行，一个命令运行时后面的事件排队。
+
+`torrent.organized` 的投递是「至少一次」：每次 AutoBangumi 重启，已整理的种子会再发布一次。脚本必须幂等。
+
+## 安装
+
+```bash
+cp -r examples/plugins/run-command config/plugins/local/
+```
+
+在「设置 → 插件」中开启「允许未签名插件」，启用并填写命令。Docker 中命令在 AutoBangumi 的容器里运行，只能使用镜像里有的程序和挂载进来的脚本。
+
+## 测试
+
+```bash
+cd examples/plugins/run-command && uv run pytest
+```
+
+对应 issue：[#897](https://github.com/EstrellaXD/Auto_Bangumi/issues/897)、[#668](https://github.com/EstrellaXD/Auto_Bangumi/issues/668)
--- a/examples/plugins/run-command/plugin.toml
+++ b/examples/plugins/run-command/plugin.toml
@@ -0,0 +1,9 @@
+[plugin]
+id = "run-command"
+name = "运行外部命令"
+version = "0.1.0"
+sdk = ">=0.5,<1"
+entry = "run_command:RunCommand"
+description = "示例插件：种子整理完成（或文件重命名）后运行一条外部命令，事件字段以环境变量和参数占位符传入。演示子进程与超时"
+authors = ["AutoBangumi"]
+permissions = ["subprocess"]
--- a/examples/plugins/run-command/pyproject.toml
+++ b/examples/plugins/run-command/pyproject.toml
@@ -0,0 +1,13 @@
+# 只用于本地开发与测试，不会被打进插件包（ab-plugin pack 排除它）
+[project]
+name = "ab-plugin-run-command"
+version = "0.1.0"
+requires-python = ">=3.14"
+dependencies = ["autobangumi-sdk"]
+
+[dependency-groups]
+dev = ["pytest>=8.0.0"]
+
+[tool.pytest.ini_options]
+pythonpath = ["."]
+testpaths = ["tests"]
--- a/examples/plugins/run-command/run_command/init.py
+++ b/examples/plugins/run-command/run_command/init.py
@@ -0,0 +1,178 @@
+"""示例插件：整理完成后运行外部命令。
+
+订阅全部事件（``@subscribe("*")``），只对配置里列出的 ``torrent.organized`` /
+``file.renamed`` 运行 ``command``。命令是参数列表，直接 ``exec``，不经过 shell，
+所以参数里的空格、``$``、``;`` 都按字面传给程序。
+
+事件字段以 ``AB_*`` 环境变量传给命令。整个参数恰好是 ``{字段}`` 时替换为字段值
+（``{files}`` 展开为每个文件一个参数）；参数里夹带的占位符不替换，以免文件名被拼进
+``sh -c`` 之类的脚本文本：
+
+- 两个事件都有：``AB_EVENT``、``AB_BANGUMI_ID``（旧种子可能为空）、``AB_DOWNLOADER_ID``
+- ``torrent.organized``：``AB_TORRENT_HASH``、``AB_FILES``（正片与字幕的路径，每行一个）
+- ``file.renamed``：``AB_OLD_PATH``、``AB_NEW_PATH``、``AB_FILE_KIND``
+
+路径是下载器视角。命令超时被杀掉；退出码非 0 时记录 stderr 的最后 2000 字节。
+命令的失败只记日志，不抛给宿主：外部脚本出错不应让插件被熔断。
+"""
+
+import asyncio
+import os
+import re
+from subprocess import DEVNULL
+
+from pydantic import BaseModel, Field
+
+from ab_sdk import Event, Plugin, subscribe
+from ab_sdk.events import FileRenamed, TorrentOrganized
+
+MAX_TIMEOUT = 3600
+STDERR_TAIL = 2000
+_PLACEHOLDER = re.compile(r"\{(\w+)\}")
+
+
+class Options(BaseModel):
+    command: list[str] = Field(
+        default_factory=list,
+        title="命令",
+        description=(
+            "程序及其参数，每项一个，如 /config/scripts/link.sh、{files}；不经过 shell，"
+            "只替换整项为占位符的参数"
+        ),
+    )
+    events: list[str] = Field(
+        default_factory=lambda: [TorrentOrganized.kind],
+        title="事件",
+        description="在哪些事件后运行：torrent.organized、file.renamed",
+    )
+    timeout: int = Field(
+        300, ge=1, le=MAX_TIMEOUT, title="超时（秒）", description="超时后杀掉命令"
+    )
+
+
+def event_fields(event: Event) -> dict[str, str] | None:
+    """事件 → 字段（键为小写，不带 ``AB_`` 前缀）；不支持的事件返回 None。"""
+    match event:
+        case TorrentOrganized():
+            specific = {
+                "torrent_hash": event.torrent_hash,
+                "files": "\n".join(f.path for f in event.files),
+            }
+        case FileRenamed():
+            specific = {
+                "old_path": event.old_path,
+                "new_path": event.new_path,
+                "file_kind": event.file_kind,
+            }
+        case _:
+            return None
+    bangumi_id = "" if event.bangumi_id is None else str(event.bangumi_id)
+    return {
+        "event": event.kind,
+        "bangumi_id": bangumi_id,
+        "downloader_id": event.downloader_id,
+        **specific,
+    }
+
+
+def build_argv(
+    command: list[str], fields: dict[str, str], files: list[str]
+) -> list[str]:
+    """整项为 ``{字段}`` 的参数替换为字段值，``{files}`` 展开为每个文件一个参数。
+
+    ``{files}`` 取原始路径列表，而不是按行拆 ``AB_FILES``：路径里的换行不会拆出
+    多余的参数。部分包含占位符的参数与未知的占位符原样保留。
+    """
+    argv: list[str] = []
+    for arg in command:
+        m = _PLACEHOLDER.fullmatch(arg)
+        if m is None or m[1] not in fields:
+            argv.append(arg)
+        elif m[1] == "files":
+            argv.extend(files)
+        else:
+            argv.append(fields[m[1]])
+    return argv
+
+
+async def drain_tail(stream: asyncio.StreamReader, tail: bytearray) -> None:
+    """持续读完 ``stream``，只保留最后 ``STDERR_TAIL`` 字节，内存有上限。"""
+    while chunk := await stream.read(65536):
+        tail += chunk
+        del tail[:-STDERR_TAIL]
+
+
+def build_env(fields: dict[str, str]) -> dict[str, str]:
+    """继承宿主环境（否则子进程没有 PATH），再加上 ``AB_*``。
+
+    宿主自己的 ``AB_*``（如 ``AB_DOWNLOADER_PASSWORD``）不传给命令。
+    """
+    env = {k: v for k, v in os.environ.items() if not k.startswith("AB_")}
+    return {**env, **{f"AB_{k.upper()}": v for k, v in fields.items()}}
+
+
+class RunCommand(Plugin[Options]):
+    config_model = Options
+
+    @subscribe("*", timeout=MAX_TIMEOUT + 10)
+    async def on_event(self, event: Event) -> None:
+        cfg = self.config
+        if not cfg.command or event.kind not in cfg.events:
+            return
+        fields = event_fields(event)
+        if fields is None:
+            return
+        files = (
+            [f.path for f in event.files] if isinstance(event, TorrentOrganized) else []
+        )
+        argv = build_argv(cfg.command, fields, files)
+        # 自建管道而不是 stderr=PIPE：读端由插件自己关闭。命令留在后台的子进程继承了
+        # stderr 时，管道要等它们退出才会关闭，不能让宿主一直挂着它
+        stream = asyncio.StreamReader()
+        r, w = os.pipe()
+        pipe, _ = await asyncio.get_running_loop().connect_read_pipe(
+            lambda: asyncio.StreamReaderProtocol(stream), os.fdopen(r, "rb")
+        )
+        try:
+            proc = await asyncio.create_subprocess_exec(
+                *argv,
+                stdin=DEVNULL,
+                stdout=DEVNULL,
+                stderr=w,
+                env=build_env(fields),
+            )
+        except OSError as e:
+            pipe.close()
+            self.ctx.log.warning("无法启动命令 %s：%s", argv[0], e)
+            return
+        finally:
+            os.close(w)
+        # 边读边丢：命令狂写 stderr 时既不会写满管道卡住，也不会占满内存或磁盘
+        tail = bytearray()
+        reader = asyncio.create_task(drain_tail(stream, tail))
+        try:
+            # 是否完成只看 proc.wait()，不等 stderr 读到 EOF
+            async with asyncio.timeout(cfg.timeout):
+                await proc.wait()
+        except TimeoutError:
+            self.ctx.log.warning("命令 %s 超过 %s 秒，已终止", argv[0], cfg.timeout)
+            return
+        finally:
+            # 超时或宿主取消处理时，不留下孤儿进程
+            # ponytail: 只杀直接子进程；命令自己再派生的进程需由脚本负责
+            if proc.returncode is None:
+                proc.kill()
+                await proc.wait()
+            # 给读取一点时间读完管道里剩下的内容，后台子进程不关管道时放弃
+            await asyncio.wait({reader}, timeout=0.5)
+            reader.cancel()
+            pipe.close()
+        if proc.returncode:
+            self.ctx.log.warning(
+                "命令 %s 退出码 %s：%s",
+                argv[0],
+                proc.returncode,
+                tail.decode(errors="replace").strip(),
+            )
+        else:
+            self.ctx.log.info("命令 %s 已完成（%s）", argv[0], event.kind)
--- a/examples/plugins/run-command/tests/test_run_command.py
+++ b/examples/plugins/run-command/tests/test_run_command.py
@@ -0,0 +1,150 @@
+import asyncio
+import json
+import logging
+import sys
+import time
+
+import pytest
+from run_command import RunCommand
+
+from ab_sdk.events import FileRenamed, OrganizedFile, TorrentOrganized
+from ab_sdk.testing import create_plugin
+
+ORGANIZED = TorrentOrganized(
+    torrent_hash="abc",
+    bangumi_id=None,
+    files=(
+        OrganizedFile("/dl/Show/Season 1/Show S01E01.mkv", "media"),
+        OrganizedFile("/dl/Show/Season 1/Show S01E01.ass", "subtitle"),
+    ),
+)
+RENAMED = FileRenamed(
+    bangumi_id=7, old_path="/dl/a.mkv", new_path="/dl/b.mkv", file_kind="media"
+)
+
+# 子进程把收到的参数与 AB_* 环境变量写进 argv[1] 指定的文件
+DUMP = (
+    "import json, os, sys; "
+    "json.dump({'argv': sys.argv[2:], "
+    "'env': {k: v for k, v in os.environ.items() if k.startswith('AB_')}}, "
+    "open(sys.argv[1], 'w'))"
+)
+
+
+def run(options: dict, event=ORGANIZED) -> None:
+    plugin, _ = create_plugin(RunCommand, options)
+    asyncio.run(plugin.on_event(event))
+
+
+def test_on_event_organized_passes_env_and_placeholders_without_shell(
+    tmp_path, monkeypatch
+):
+    # 宿主自己的 AB_* 凭据不传给命令
+    monkeypatch.setenv("AB_DOWNLOADER_PASSWORD", "s3cret")
+    out = tmp_path / "out.json"
+    command = [
+        *(sys.executable, "-c", DUMP, str(out)),
+        *("{torrent_hash}", "$HOME; {x}", "pre-{torrent_hash}", "{files}"),
+    ]
+    run({"command": command})
+    got = json.loads(out.read_text())
+    # $HOME 与 ; 按字面传入：没有经过 shell；未知占位符与夹在参数里的占位符原样保留；
+    # {files} 展开为每个文件一个参数
+    assert got["argv"] == [
+        "abc",
+        "$HOME; {x}",
+        "pre-{torrent_hash}",
+        "/dl/Show/Season 1/Show S01E01.mkv",
+        "/dl/Show/Season 1/Show S01E01.ass",
+    ]
+    assert got["env"] == {
+        "AB_EVENT": "torrent.organized",
+        "AB_BANGUMI_ID": "",
+        "AB_DOWNLOADER_ID": "default",
+        "AB_TORRENT_HASH": "abc",
+        "AB_FILES": "/dl/Show/Season 1/Show S01E01.mkv\n"
+        "/dl/Show/Season 1/Show S01E01.ass",
+    }
+
+
+def test_on_event_renamed_when_selected_passes_rename_fields(tmp_path):
+    out = tmp_path / "out.json"
+    command = [sys.executable, "-c", DUMP, str(out), "{old_path}"]
+    run({"command": command, "events": ["file.renamed"]}, RENAMED)
+    got = json.loads(out.read_text())
+    assert got["argv"] == ["/dl/a.mkv"]
+    assert got["env"]["AB_BANGUMI_ID"] == "7"
+    assert got["env"]["AB_NEW_PATH"] == "/dl/b.mkv"
+    assert got["env"]["AB_FILE_KIND"] == "media"
+
+
+@pytest.mark.parametrize(
+    ("options", "event"),
+    [({"command": []}, ORGANIZED), ({}, RENAMED)],
+)
+def test_on_event_unwanted_runs_nothing(tmp_path, options, event):
+    out = tmp_path / "out.json"
+    options.setdefault("command", [sys.executable, "-c", DUMP, str(out)])
+    run(options, event)
+    assert not out.exists()
+
+
+def test_on_event_timeout_kills_command(caplog):
+    command = [sys.executable, "-c", "import time; time.sleep(30)"]
+    plugin, _ = create_plugin(RunCommand, {"command": command, "timeout": 1})
+    start = time.monotonic()
+    with caplog.at_level(logging.WARNING):
+        asyncio.run(plugin.on_event(ORGANIZED))
+    assert time.monotonic() - start < 10
+    assert "已终止" in caplog.text
+
+
+def test_on_event_background_child_does_not_delay_success(caplog):
+    # 后台进程继承了 stderr；命令本身已退出，不应等到后台进程结束
+    command = ["sh", "-c", "sleep 5 & exit 0"]
+    plugin, _ = create_plugin(RunCommand, {"command": command, "timeout": 2})
+    start = time.monotonic()
+    with caplog.at_level(logging.INFO):
+        asyncio.run(plugin.on_event(ORGANIZED))
+    assert time.monotonic() - start < 2
+    assert "已完成" in caplog.text and "已终止" not in caplog.text
+
+
+def test_on_event_files_placeholder_keeps_newline_in_path(tmp_path):
+    # {files} 取原始路径列表：含换行的路径仍是一个参数（AB_FILES 只能按行拼接）
+    out = tmp_path / "out.json"
+    event = TorrentOrganized(
+        torrent_hash="abc",
+        bangumi_id=1,
+        files=(OrganizedFile("/dl/a\nb.mkv", "media"),),
+    )
+    run({"command": [sys.executable, "-c", DUMP, str(out), "{files}"]}, event)
+    assert json.loads(out.read_text())["argv"] == ["/dl/a\nb.mkv"]
+
+
+def test_on_event_huge_stderr_keeps_only_tail(caplog):
+    # 输出远超管道缓冲：边读边丢，命令不会卡到超时，日志只带最后 2000 字节
+    script = (
+        "import sys; sys.stderr.write('x' * 5_000_000 + 'END'); sys.stderr.flush(); "
+        "sys.exit(3)"
+    )
+    start = time.monotonic()
+    with caplog.at_level(logging.WARNING):
+        run({"command": [sys.executable, "-c", script], "timeout": 10})
+    assert time.monotonic() - start < 5
+    [record] = caplog.records
+    assert "退出码 3" in record.getMessage()
+    assert record.args[-1] == "x" * 1997 + "END"
+
+
+def test_on_event_nonzero_exit_logs_code_and_stderr(caplog):
+    script = "import sys; sys.stderr.write('boom'); sys.exit(3)"
+    with caplog.at_level(logging.WARNING):
+        run({"command": [sys.executable, "-c", script]})
+    assert "退出码 3" in caplog.text and "boom" in caplog.text
+
+
+def test_on_event_missing_program_is_logged_not_raised(caplog):
+    with caplog.at_level(logging.WARNING):
+        run({"command": ["/nonexistent/ab-run-command-test"]})
+    assert "无法启动命令" in caplog.text
--- a/examples/plugins/slack-notifier/README.md
+++ b/examples/plugins/slack-notifier/README.md
@@ -0,0 +1,28 @@
+# Slack 通知（示例插件）
+
+把通知发到 Slack 的 [Incoming Webhook](https://api.slack.com/messaging/webhooks)。演示：
+
+- `@provider(points.NOTIFIER, id="slack")`：返回 `NotifierFactory`
+- Webhook 地址含密钥，用 `secret_field` 声明，读取配置时只返回掩码
+- 只有 HTTP 2xx 且响应体为 `ok` 才算成功
+- 契约套件 `NotifierContract`
+
+## 消息格式
+
+请求体为 `{"text": "*标题*\n正文"}`（Slack 的 mrkdwn，`&`、`<`、`>` 已转义）。配置了 `notification.base_url` 时单集通知带海报：再加 `blocks`（一段文字和一张图片），`text` 作为手机通知的摘要。海报地址必须能被 Slack 访问；Slack 取不到图片时会以 HTTP 400 `invalid_blocks` 拒绝整条消息，插件只在这种情况下去掉 `blocks` 只发文字再试一次；令牌失效、限流等其它失败不重试，避免重复发送。
+
+## 安装
+
+```bash
+cp -r examples/plugins/slack-notifier config/plugins/local/
+```
+
+在 Slack 中创建 App，开启 Incoming Webhooks 并添加到频道，复制 Webhook 地址。在「设置 → 插件」中开启「允许未签名插件」，启用并填写地址；再到「设置 → 通知 → 添加渠道」选择类型 `slack`。
+
+## 测试
+
+```bash
+cd examples/plugins/slack-notifier && uv run pytest
+```
+
+对应 issue：[#277](https://github.com/EstrellaXD/Auto_Bangumi/issues/277)
--- a/examples/plugins/slack-notifier/plugin.toml
+++ b/examples/plugins/slack-notifier/plugin.toml
@@ -0,0 +1,10 @@
+[plugin]
+id = "slack-notifier"
+name = "Slack 通知"
+version = "0.1.0"
+sdk = ">=0.5,<1"
+entry = "slack_notifier:SlackNotifier"
+description = "示例插件：把通知发到 Slack 的 Incoming Webhook，带海报时附图片。在「设置 → 通知 → 添加渠道」选择类型 slack"
+authors = ["AutoBangumi"]
+permissions = ["network"]
+extension_points = ["notifier"]
--- a/examples/plugins/slack-notifier/pyproject.toml
+++ b/examples/plugins/slack-notifier/pyproject.toml
@@ -0,0 +1,13 @@
+# 只用于本地开发与测试，不会被打进插件包（ab-plugin pack 排除它）
+[project]
+name = "ab-plugin-slack-notifier"
+version = "0.1.0"
+requires-python = ">=3.14"
+dependencies = ["autobangumi-sdk"]
+
+[dependency-groups]
+dev = ["pytest>=8.0.0"]
+
+[tool.pytest.ini_options]
+pythonpath = ["."]
+testpaths = ["tests"]
--- a/examples/plugins/slack-notifier/slack_notifier/init.py
+++ b/examples/plugins/slack-notifier/slack_notifier/init.py
@@ -0,0 +1,99 @@
+"""示例插件：Slack 通知渠道。
+
+向 Incoming Webhook POST ``{"text": "*标题*\\n正文"}``（Slack 的 mrkdwn）。带海报时
+再加 ``blocks``：一段文字加一张图片，``text`` 留作手机通知里的摘要。Slack 成功时
+返回 HTTP 200 与响应体 ``ok``，失败时返回 4xx 与错误码（如 ``invalid_token``）。
+Slack 取不到海报图片时会拒绝整条消息（HTTP 400 ``invalid_blocks``），只有这种情况
+才去掉 ``blocks`` 只发文字再试一次；令牌失效、限流等其它失败不重试，以免重复发送。
+"""
+
+from collections.abc import Callable
+from typing import Any
+
+import httpx2
+from pydantic import BaseModel
+
+from ab_sdk import Plugin, points, provider, secret_field
+from ab_sdk.notify import NotificationMessage, NotifierFactory, NotifierSettings
+
+
+class Options(BaseModel):
+    webhook_url: str = secret_field(
+        title="Webhook 地址",
+        description="Slack App 的 Incoming Webhook 地址，形如 https://hooks.slack.com/services/...",
+    )
+
+
+def escape(text: str) -> str:
+    """mrkdwn 只要求转义这三个字符（番剧名里常有 &）。"""
+    return text.replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")
+
+
+def slack_body(message: NotificationMessage) -> dict[str, Any]:
+    text = f"*{escape(message.title)}*\n{escape(message.body)}"
+    body: dict[str, Any] = {"text": text}
+    if message.poster_url:
+        body["blocks"] = [
+            {"type": "section", "text": {"type": "mrkdwn", "text": text}},
+            {
+                "type": "image",
+                "image_url": message.poster_url,
+                "alt_text": message.title,
+            },
+        ]
+    return body
+
+
+def _ok(resp: httpx2.Response) -> bool:
+    return resp.is_success and resp.text == "ok"
+
+
+def _blocks_rejected(resp: httpx2.Response) -> bool:
+    """Slack 因 blocks（如下载不到图片）拒绝消息；``invalid_blocks_format`` 同属此类。"""
+    return resp.status_code == 400 and "invalid_blocks" in resp.text
+
+
+class Channel:
+    def __init__(
+        self,
+        options: Options,
+        settings: NotifierSettings,
+        client_factory: Callable[[], httpx2.AsyncClient],
+    ) -> None:
+        self.options = options
+        self.settings = settings
+        self.client_factory = client_factory
+
+    async def send(self, message: NotificationMessage) -> bool:
+        if not self.options.webhook_url:
+            return False
+        body = slack_body(message)
+        try:
+            async with self.client_factory() as client:
+                resp = await client.post(self.options.webhook_url, json=body)
+                if "blocks" in body and _blocks_rejected(resp):
+                    resp = await client.post(
+                        self.options.webhook_url, json={"text": body["text"]}
+                    )
+        except httpx2.HTTPError:
+            return False
+        return _ok(resp)
+
+    async def test(self) -> tuple[bool, str]:
+        ok = await self.send(
+            NotificationMessage(kind="event", title="AutoBangumi", body="通知测试")
+        )
+        return ok, "ok" if ok else "发送失败"
+
+
+class SlackNotifier(Plugin[Options]):
+    config_model = Options
+
+    # 测试可替换为带 MockTransport 的客户端
+    client_factory: Callable[[], httpx2.AsyncClient] = staticmethod(
+        lambda: httpx2.AsyncClient(timeout=10)
+    )
+
+    @provider(points.NOTIFIER, id="slack")
+    def channel(self) -> NotifierFactory:
+        return lambda settings: Channel(self.config, settings, self.client_factory)
--- a/examples/plugins/slack-notifier/tests/test_slack_notifier.py
+++ b/examples/plugins/slack-notifier/tests/test_slack_notifier.py
@@ -0,0 +1,109 @@
+import asyncio
+import json
+from collections.abc import Callable
+
+import httpx2
+import pytest
+from slack_notifier import SlackNotifier
+
+from ab_sdk.notify import NotificationMessage, NotifierSettings
+from ab_sdk.testing import NotifierContract, create_plugin
+
+URL = "https://hooks.slack.com/services/T/B/x"
+
+
+def channel(
+    response: httpx2.Response | Callable[[dict], httpx2.Response],
+    seen: list | None = None,
+):
+    plugin, _ = create_plugin(SlackNotifier, {"webhook_url": URL})
+
+    def handler(request: httpx2.Request) -> httpx2.Response:
+        if seen is not None:
+            seen.append(request)
+        if callable(response):
+            return response(json.loads(request.content))
+        return response
+
+    plugin.client_factory = lambda: httpx2.AsyncClient(  # type: ignore[method-assign]
+        transport=httpx2.MockTransport(handler)
+    )
+    return plugin.channel()(NotifierSettings())
+
+
+class TestChannel(NotifierContract):
+    def create(self):
+        return channel(httpx2.Response(200, text="ok"))
+
+    def create_failing(self):
+        return channel(httpx2.Response(403, text="invalid_token"))
+
+
+def sent_body(message: NotificationMessage) -> dict:
+    seen: list[httpx2.Request] = []
+    assert asyncio.run(channel(httpx2.Response(200, text="ok"), seen).send(message))
+    return json.loads(seen[0].content)
+
+
+def test_send_escapes_mrkdwn_and_bolds_title():
+    body = sent_body(NotificationMessage(kind="event", title="A & <B>", body="x"))
+    assert body == {"text": "*A &amp; &lt;B&gt;*\nx"}
+
+
+def test_send_with_poster_adds_image_block():
+    message = NotificationMessage(
+        kind="episode", title="T", body="b", poster_url="https://p/1.jpg"
+    )
+    body = sent_body(message)
+    assert body["text"] == "*T*\nb"
+    assert body["blocks"][1] == {
+        "type": "image",
+        "image_url": "https://p/1.jpg",
+        "alt_text": "T",
+    }
+
+
+@pytest.mark.parametrize("text", ["", "no_service"])
+def test_send_2xx_without_ok_body_is_failure(text):
+    message = NotificationMessage(kind="event", title="t", body="b")
+    assert asyncio.run(channel(httpx2.Response(200, text=text)).send(message)) is False
+
+
+# Slack 取不到海报时拒绝带 blocks 的整条消息
+def reject_blocks(body: dict) -> httpx2.Response:
+    if "blocks" in body:
+        return httpx2.Response(400, text="invalid_blocks")
+    return httpx2.Response(200, text="ok")
+
+
+def test_send_rejected_blocks_retries_once_with_text_only():
+    seen: list[httpx2.Request] = []
+    message = NotificationMessage(
+        kind="episode", title="T", body="b", poster_url="https://p/1.jpg"
+    )
+    assert asyncio.run(channel(reject_blocks, seen).send(message)) is True
+    first, retry = seen
+    assert "blocks" in json.loads(first.content)
+    assert json.loads(retry.content) == {"text": "*T*\nb"}
+
+
+def test_send_text_only_failure_is_not_retried():
+    seen: list[httpx2.Request] = []
+    message = NotificationMessage(kind="event", title="t", body="b")
+    ch = channel(httpx2.Response(400, text="invalid_payload"), seen)
+    assert asyncio.run(ch.send(message)) is False
+    assert len(seen) == 1
+
+
+@pytest.mark.parametrize(
+    ("status", "text"), [(403, "invalid_token"), (429, "rate_limited")]
+)
+def test_send_with_poster_other_failure_is_not_retried(status, text):
+    # 只有 blocks 被拒才退回纯文字；其它失败重发可能重复投递
+    seen: list[httpx2.Request] = []
+    message = NotificationMessage(
+        kind="episode", title="T", body="b", poster_url="https://p/1.jpg"
+    )
+    ch = channel(httpx2.Response(status, text=text), seen)
+    assert asyncio.run(ch.send(message)) is False
+    assert len(seen) == 1
--- a/examples/plugins/strm-writer/README.md
+++ b/examples/plugins/strm-writer/README.md
@@ -0,0 +1,35 @@
+# STRM 写入（示例插件）
+
+种子整理完成后，为每个正片写一个 `.strm`。把 STRM 目录加入 Emby / Jellyfin 媒体库，再把媒体文件放到网盘或对象存储，客户端播放时直接从该地址读取，不经过媒体服务器。演示：
+
+- `@subscribe(TorrentOrganized.kind)`：只处理 `kind == "media"` 的文件
+- 投递为「至少一次」：内容相同就不重写，`.strm` 的修改时间不变
+- 文件操作放进 `asyncio.to_thread`，不阻塞事件循环
+
+## 配置
+
+| 字段 | 示例 | 说明 |
+| --- | --- | --- |
+| 媒体根目录 | `/downloads/Bangumi` | 下载器中要去掉的路径前缀（与事件路径同为下载器视角）。不在其下的文件被跳过并记录警告；填 `/` 时保留完整路径 |
+| STRM 目录 | `/downloads/STRM` | 写入 `.strm` 的目录，是 AutoBangumi 看到的路径 |
+| 地址前缀 | `http://nas:8096/media/` | 拼在相对路径前。含 `://` 时，路径的每一段做百分号编码（空格变为 `%20`）；不含时（如本地路径 `/mnt/cloud/`）原样拼接 |
+
+例：正片 `/downloads/Bangumi/A/Season 1/A S01E01.mkv`，按上表配置写入 `/downloads/STRM/A/Season 1/A S01E01.strm`，内容为 `http://nas:8096/media/A/Season%201/A%20S01E01.mkv`。
+
+前缀以 `/` 结尾时直接拼接相对路径；插件不补 `/`。字幕不写 `.strm`。
+
+## 安装
+
+```bash
+cp -r examples/plugins/strm-writer config/plugins/local/
+```
+
+在「设置 → 插件」中开启「允许未签名插件」，启用并填写三个字段。
+
+## 测试
+
+```bash
+cd examples/plugins/strm-writer && uv run pytest
+```
+
+对应 issue：[#428](https://github.com/EstrellaXD/Auto_Bangumi/issues/428)
--- a/examples/plugins/strm-writer/plugin.toml
+++ b/examples/plugins/strm-writer/plugin.toml
@@ -0,0 +1,9 @@
+[plugin]
+id = "strm-writer"
+name = "STRM 写入"
+version = "0.1.0"
+sdk = ">=0.5,<1"
+entry = "strm_writer:StrmWriter"
+description = "示例插件：种子整理完成后，按媒体目录结构在 STRM 目录写 .strm，内容为地址前缀加相对路径。演示 torrent.organized 事件与幂等写入"
+authors = ["AutoBangumi"]
+permissions = ["fs.write"]
--- a/examples/plugins/strm-writer/pyproject.toml
+++ b/examples/plugins/strm-writer/pyproject.toml
@@ -0,0 +1,13 @@
+# 只用于本地开发与测试，不会被打进插件包（ab-plugin pack 排除它）
+[project]
+name = "ab-plugin-strm-writer"
+version = "0.1.0"
+requires-python = ">=3.14"
+dependencies = ["autobangumi-sdk"]
+
+[dependency-groups]
+dev = ["pytest>=8.0.0"]
+
+[tool.pytest.ini_options]
+pythonpath = ["."]
+testpaths = ["tests"]
--- a/examples/plugins/strm-writer/strm_writer/init.py
+++ b/examples/plugins/strm-writer/strm_writer/init.py
@@ -0,0 +1,89 @@
+"""示例插件：STRM 写入。
+
+订阅 ``torrent.organized``。对事件里每个正片，去掉路径前缀 ``media_root`` 得到
+相对路径 ``rel``，写 ``<strm_root>/<rel 换成 .strm>``，内容为 ``url_prefix + rel``。
+Emby / Jellyfin 播放 ``.strm`` 时直接读取这个地址，媒体文件可以放在网盘或对象存储里。
+
+- ``media_root`` 与事件路径同为下载器视角，所以不需要路径映射；``strm_root`` 是
+  AutoBangumi 看到的本地目录。
+- ``url_prefix`` 含 ``://`` 时按 URL 处理，路径的每一段做百分号编码；否则
+  （本地路径）原样拼接。
+- 投递为「至少一次」：内容相同就不重写；``url_prefix`` 改了会在下次事件时更新。
+"""
+
+import asyncio
+import posixpath
+from pathlib import Path
+from urllib.parse import quote
+
+from pydantic import BaseModel, Field
+
+from ab_sdk import Plugin, subscribe
+from ab_sdk.events import TorrentOrganized
+
+
+class Options(BaseModel):
+    strm_root: str = Field(
+        "", title="STRM 目录", description="写入 .strm 的目录（AutoBangumi 中的路径）"
+    )
+    url_prefix: str = Field(
+        "",
+        title="地址前缀",
+        description="拼在相对路径前，如 http://nas:8096/media/ 或本地路径 /mnt/cloud/",
+    )
+    media_root: str = Field(
+        "",
+        title="媒体根目录",
+        description="下载器中要去掉的路径前缀，如 /downloads/Bangumi；其下的目录结构原样映射",
+    )
+
+
+def relative(path: str, media_root: str) -> str | None:
+    """``path`` 不在 ``media_root`` 下（或就是 ``media_root`` 本身）时返回 None。"""
+    # media_root 为 "/" 时 root 为空串，前缀 "/" 照样匹配所有绝对路径
+    root = media_root.rstrip("/")
+    if not media_root or not path.startswith(root + "/"):
+        return None
+    rel = posixpath.normpath(path[len(root) + 1 :])
+    # "." 会写出 <strm_root>.strm，落在 strm_root 之外
+    if rel in (".", "..") or rel.startswith("../"):
+        return None
+    return rel
+
+
+def strm_target(rel: str, options: Options) -> tuple[Path, str]:
+    """(.strm 路径, 文件内容)。"""
+    link = quote(rel, safe="/") if "://" in options.url_prefix else rel
+    strm = Path(options.strm_root, *rel.split("/")).with_suffix(".strm")
+    return strm, options.url_prefix + link
+
+
+def write_strm(strm: Path, content: str) -> bool:
+    """内容相同时不重写，返回 False。"""
+    if strm.is_file() and strm.read_text(encoding="utf-8") == content:
+        return False
+    strm.parent.mkdir(parents=True, exist_ok=True)
+    strm.write_text(content, encoding="utf-8")
+    return True
+
+
+class StrmWriter(Plugin[Options]):
+    config_model = Options
+
+    @subscribe(TorrentOrganized.kind)
+    async def on_organized(self, event: TorrentOrganized) -> None:
+        cfg = self.config
+        if not cfg.strm_root or not cfg.url_prefix:
+            return
+        for file in event.files:
+            if file.kind != "media":
+                continue
+            rel = relative(file.path, cfg.media_root)
+            if rel is None:
+                self.ctx.log.warning(
+                    "%s 不在媒体根目录 %s 下", file.path, cfg.media_root
+                )
+                continue
+            strm, content = strm_target(rel, cfg)
+            if await asyncio.to_thread(write_strm, strm, content):
+                self.ctx.log.info("已写入 %s", strm)
--- a/examples/plugins/strm-writer/tests/test_strm_writer.py
+++ b/examples/plugins/strm-writer/tests/test_strm_writer.py
@@ -0,0 +1,82 @@
+import asyncio
+import logging
+
+import pytest
+from strm_writer import StrmWriter
+
+from ab_sdk.events import OrganizedFile, TorrentOrganized
+from ab_sdk.testing import create_plugin
+
+MEDIA = "/downloads/Bangumi/A/Season 1/A S01E01.mkv"
+
+
+def organized(*files: OrganizedFile) -> TorrentOrganized:
+    return TorrentOrganized(torrent_hash="h", bangumi_id=1, files=files)
+
+
+def run(
+    tmp_path,
+    url_prefix: str,
+    *files: OrganizedFile,
+    media_root: str = "/downloads/Bangumi/",
+) -> None:
+    options = {
+        "strm_root": str(tmp_path / "STRM"),
+        "url_prefix": url_prefix,
+        "media_root": media_root,
+    }
+    plugin, _ = create_plugin(StrmWriter, options)
+    asyncio.run(plugin.on_organized(organized(*files)))
+
+
+@pytest.mark.parametrize(
+    ("url_prefix", "content"),
+    [
+        (
+            "http://nas:8096/media/",
+            "http://nas:8096/media/A/Season%201/A%20S01E01.mkv",
+        ),
+        ("/mnt/cloud/", "/mnt/cloud/A/Season 1/A S01E01.mkv"),
+    ],
+)
+def test_on_organized_mirrors_media_path_into_strm(tmp_path, url_prefix, content):
+    run(
+        tmp_path,
+        url_prefix,
+        OrganizedFile(MEDIA, "media"),
+        OrganizedFile(MEDIA.replace(".mkv", ".ass"), "subtitle"),
+    )
+    season = tmp_path / "STRM" / "A" / "Season 1"
+    assert [p.name for p in season.iterdir()] == ["A S01E01.strm"]
+    assert (season / "A S01E01.strm").read_text(encoding="utf-8") == content
+
+
+def test_on_organized_media_root_slash_maps_whole_path(tmp_path):
+    # 根目录 "/" 去掉尾部斜杠后是空串，曾把所有绝对路径都判为不在根目录下
+    run(tmp_path, "/mnt/", OrganizedFile("/dl/A/B.mkv", "media"), media_root="/")
+    strm = tmp_path / "STRM" / "dl" / "A" / "B.strm"
+    assert strm.read_text(encoding="utf-8") == "/mnt/dl/A/B.mkv"
+
+
+def test_on_organized_prefix_change_rewrites_strm(tmp_path):
+    run(tmp_path, "http://old/", OrganizedFile(MEDIA, "media"))
+    run(tmp_path, "http://new/", OrganizedFile(MEDIA, "media"))
+    strm = tmp_path / "STRM" / "A" / "Season 1" / "A S01E01.strm"
+    assert strm.read_text(encoding="utf-8").startswith("http://new/")
+
+
+@pytest.mark.parametrize(
+    "path",
+    [
+        "/other/A S01E01.mkv",
+        "/downloads/Bangumi/../../etc/x.mkv",
+        # 规范化后是媒体根目录本身，曾写出 STRM 目录之外的 STRM.strm
+        "/downloads/Bangumi/x/..",
+        "/downloads/Bangumi/",
+    ],
+)
+def test_on_organized_path_outside_media_root_is_skipped(tmp_path, caplog, path):
+    with caplog.at_level(logging.WARNING):
+        run(tmp_path, "http://nas/", OrganizedFile(path, "media"))
+    assert list(tmp_path.iterdir()) == []
+    assert "不在媒体根目录" in caplog.text
--- a/examples/plugins/synology-chat-notifier/README.md
+++ b/examples/plugins/synology-chat-notifier/README.md
@@ -0,0 +1,30 @@
+# Synology Chat 通知（示例插件）
+
+把通知发到群晖 Synology Chat 的传入 Webhook。演示：
+
+- `@provider(points.NOTIFIER, id="synology-chat")`：返回 `NotifierFactory`
+- Webhook 地址含 token，用 `secret_field` 声明，读取配置时只返回掩码
+- 后端返回 HTTP 200 但 `{"success": false}` 时，`send` 同样返回 `False`
+- 契约套件 `NotifierContract`
+
+## 获取 Webhook 地址
+
+在 Synology Chat 中打开「整合 → 传入 Webhook」，新建一个并复制地址（形如 `https://<nas>:5001/webapi/entry.cgi?api=SYNO.Chat.External&method=incoming&version=2&token=...`）。
+
+请求为表单字段 `payload=<JSON>`，JSON 为 `{"text": "标题\n正文"}`；配置了 `notification.base_url` 时单集通知带海报地址 `file_url`，Chat 会下载并显示该图片，所以海报地址必须能被 NAS 访问。
+
+## 安装
+
+```bash
+cp -r examples/plugins/synology-chat-notifier config/plugins/local/
+```
+
+在「设置 → 插件」中开启「允许未签名插件」，启用并填写 Webhook 地址；再到「设置 → 通知 → 添加渠道」选择类型 `synology-chat`。
+
+## 测试
+
+```bash
+cd examples/plugins/synology-chat-notifier && uv run pytest
+```
+
+对应 issue：[#887](https://github.com/EstrellaXD/Auto_Bangumi/issues/887)
--- a/examples/plugins/synology-chat-notifier/plugin.toml
+++ b/examples/plugins/synology-chat-notifier/plugin.toml
@@ -0,0 +1,10 @@
+[plugin]
+id = "synology-chat-notifier"
+name = "Synology Chat 通知"
+version = "0.1.0"
+sdk = ">=0.5,<1"
+entry = "synology_chat_notifier:SynologyChatNotifier"
+description = "示例插件：把通知发到 Synology Chat 的传入 Webhook。在「设置 → 通知 → 添加渠道」选择类型 synology-chat"
+authors = ["AutoBangumi"]
+permissions = ["network"]
+extension_points = ["notifier"]
--- a/examples/plugins/synology-chat-notifier/pyproject.toml
+++ b/examples/plugins/synology-chat-notifier/pyproject.toml
@@ -0,0 +1,13 @@
+# 只用于本地开发与测试，不会被打进插件包（ab-plugin pack 排除它）
+[project]
+name = "ab-plugin-synology-chat-notifier"
+version = "0.1.0"
+requires-python = ">=3.14"
+dependencies = ["autobangumi-sdk"]
+
+[dependency-groups]
+dev = ["pytest>=8.0.0"]
+
+[tool.pytest.ini_options]
+pythonpath = ["."]
+testpaths = ["tests"]
--- a/examples/plugins/synology-chat-notifier/synology_chat_notifier/init.py
+++ b/examples/plugins/synology-chat-notifier/synology_chat_notifier/init.py
@@ -0,0 +1,78 @@
+"""示例插件：Synology Chat 通知渠道。
+
+向 Chat 的传入 Webhook 发送表单字段 ``payload=<JSON>``（Synology Chat 只接受这种
+格式），JSON 为 ``{"text": ...}``，带海报时加 ``file_url``。Chat 即使拒绝也常返回
+HTTP 200，结果在响应体的 ``success`` 里，所以两者都要检查。
+"""
+
+import json
+from collections.abc import Callable
+
+import httpx2
+from pydantic import BaseModel
+
+from ab_sdk import Plugin, points, provider, secret_field
+from ab_sdk.notify import NotificationMessage, NotifierFactory, NotifierSettings
+
+
+class Options(BaseModel):
+    webhook_url: str = secret_field(
+        title="Webhook 地址",
+        description="Chat → 整合 → 传入 Webhook 中的地址，含 token",
+    )
+
+
+def accepted(resp: httpx2.Response) -> bool:
+    """2xx，且响应体是 JSON 时其中的 success 不为 false。"""
+    if not resp.is_success:
+        return False
+    try:
+        body = resp.json()
+    except ValueError:
+        return True
+    return not isinstance(body, dict) or body.get("success", True) is not False
+
+
+class Channel:
+    def __init__(
+        self,
+        options: Options,
+        settings: NotifierSettings,
+        client_factory: Callable[[], httpx2.AsyncClient],
+    ) -> None:
+        self.options = options
+        self.settings = settings
+        self.client_factory = client_factory
+
+    async def send(self, message: NotificationMessage) -> bool:
+        if not self.options.webhook_url:
+            return False
+        payload: dict[str, str] = {"text": f"{message.title}\n{message.body}"}
+        if message.poster_url:
+            payload["file_url"] = message.poster_url
+        data = {"payload": json.dumps(payload, ensure_ascii=False)}
+        try:
+            async with self.client_factory() as client:
+                resp = await client.post(self.options.webhook_url, data=data)
+        except httpx2.HTTPError:
+            return False
+        return accepted(resp)
+
+    async def test(self) -> tuple[bool, str]:
+        ok = await self.send(
+            NotificationMessage(kind="event", title="AutoBangumi", body="通知测试")
+        )
+        return ok, "ok" if ok else "发送失败"
+
+
+class SynologyChatNotifier(Plugin[Options]):
+    config_model = Options
+
+    # 测试可替换为带 MockTransport 的客户端
+    client_factory: Callable[[], httpx2.AsyncClient] = staticmethod(
+        lambda: httpx2.AsyncClient(timeout=10)
+    )
+
+    @provider(points.NOTIFIER, id="synology-chat")
+    def channel(self) -> NotifierFactory:
+        return lambda settings: Channel(self.config, settings, self.client_factory)
--- a/examples/plugins/synology-chat-notifier/tests/test_synology_chat_notifier.py
+++ b/examples/plugins/synology-chat-notifier/tests/test_synology_chat_notifier.py
@@ -0,0 +1,63 @@
+import asyncio
+import json
+from urllib.parse import parse_qs
+
+import httpx2
+import pytest
+from synology_chat_notifier import SynologyChatNotifier
+
+from ab_sdk.notify import NotificationMessage, NotifierSettings
+from ab_sdk.testing import NotifierContract, create_plugin
+
+URL = "https://nas:5001/webapi/entry.cgi?api=SYNO.Chat.External&token=t"
+
+
+def channel(response: httpx2.Response, seen: list | None = None):
+    plugin, _ = create_plugin(SynologyChatNotifier, {"webhook_url": URL})
+
+    def handler(request: httpx2.Request) -> httpx2.Response:
+        if seen is not None:
+            seen.append(request)
+        return response
+
+    plugin.client_factory = lambda: httpx2.AsyncClient(  # type: ignore[method-assign]
+        transport=httpx2.MockTransport(handler)
+    )
+    return plugin.channel()(NotifierSettings())
+
+
+class TestChannel(NotifierContract):
+    def create(self):
+        return channel(httpx2.Response(200, json={"success": True}))
+
+    def create_failing(self):
+        # Chat 拒绝请求时仍返回 HTTP 200
+        return channel(httpx2.Response(200, json={"success": False, "error": {}}))
+
+
+def test_send_posts_form_encoded_payload_json():
+    seen: list[httpx2.Request] = []
+    ch = channel(httpx2.Response(200, json={"success": True}), seen)
+    message = NotificationMessage(
+        kind="episode", title="新番更新", body="第 5 集", poster_url="https://p/1.jpg"
+    )
+    assert asyncio.run(ch.send(message)) is True
+    [request] = seen
+    assert str(request.url) == URL
+    form = parse_qs(request.content.decode())
+    assert json.loads(form["payload"][0]) == {
+        "text": "新番更新\n第 5 集",
+        "file_url": "https://p/1.jpg",
+    }
+
+
+@pytest.mark.parametrize(
+    ("response", "ok"),
+    [
+        (httpx2.Response(200, text="ok"), True),
+        (httpx2.Response(500, json={"success": True}), False),
+    ],
+)
+def test_send_judges_status_and_optional_json_body(response, ok):
+    message = NotificationMessage(kind="event", title="t", body="b")
+    assert asyncio.run(channel(response).send(message)) is ok
--- a/examples/plugins/wpush-notifier/README.md
+++ b/examples/plugins/wpush-notifier/README.md
@@ -0,0 +1,33 @@
+# WPUSH 通知
+
+把通知发到 [WPUSH](https://wpush.cn)，支持微信、短信、邮件等渠道。
+
+- `@provider(points.NOTIFIER, id="wpush")`：返回 `NotifierFactory`
+- 向 `https://api.wpush.cn/api/v1/send` POST JSON：`apikey`、`title`、`content`、`channel`，填写了群组编码时附带 `topic_code`
+- HTTP 2xx 且返回的 `code` 为 0 才算成功；否则 `send` 返回 `False`，「测试」显示 WPUSH 返回的错误信息
+
+## 配置
+
+| 字段 | 说明 |
+| --- | --- |
+| `apikey` | WPUSH 控制台中的 apikey（秘密字段） |
+| `channel` | 推送渠道，默认 `wechat` |
+| `topic_code` | 可选：群组编码 |
+
+## 安装
+
+```bash
+cp -r examples/plugins/wpush-notifier config/plugins/local/
+```
+
+在「设置 → 插件」中开启「允许未签名插件」，启用并填写 apikey；再到「设置 → 通知 → 添加渠道」选择类型 `wpush`。
+
+## 测试
+
+```bash
+cd examples/plugins/wpush-notifier && uv run pytest
+```
+
+---
+
+基于 @anhao 在 [#1116](https://github.com/EstrellaXD/Auto_Bangumi/pull/1116) 中的实现
--- a/examples/plugins/wpush-notifier/plugin.toml
+++ b/examples/plugins/wpush-notifier/plugin.toml
@@ -0,0 +1,10 @@
+[plugin]
+id = "wpush-notifier"
+name = "WPUSH 通知"
+version = "0.1.0"
+sdk = ">=0.5,<1"
+entry = "wpush_notifier:WPushNotifier"
+description = "把通知发到 WPUSH（微信、短信、邮件等多渠道）。在「设置 → 通知 → 添加渠道」选择类型 wpush"
+authors = ["AutoBangumi"]
+permissions = ["network"]
+extension_points = ["notifier"]
--- a/examples/plugins/wpush-notifier/pyproject.toml
+++ b/examples/plugins/wpush-notifier/pyproject.toml
@@ -0,0 +1,13 @@
+# 只用于本地开发与测试，不会被打进插件包（ab-plugin pack 排除它）
+[project]
+name = "ab-plugin-wpush-notifier"
+version = "0.1.0"
+requires-python = ">=3.14"
+dependencies = ["autobangumi-sdk"]
+
+[dependency-groups]
+dev = ["pytest>=8.0.0"]
+
+[tool.pytest.ini_options]
+pythonpath = ["."]
+testpaths = ["tests"]
--- a/examples/plugins/wpush-notifier/tests/test_wpush_notifier.py
+++ b/examples/plugins/wpush-notifier/tests/test_wpush_notifier.py
@@ -0,0 +1,78 @@
+import asyncio
+import json
+
+import httpx2
+import pytest
+from wpush_notifier import API_URL, WPushNotifier
+
+from ab_sdk.notify import NotificationMessage, NotifierSettings
+from ab_sdk.testing import NotifierContract, create_plugin
+
+
+def channel(status: int, body: dict | str, options: dict | None = None, seen=None):
+    def handler(request: httpx2.Request) -> httpx2.Response:
+        if seen is not None:
+            seen.append(request)
+        if isinstance(body, str):
+            return httpx2.Response(status, text=body)
+        return httpx2.Response(status, json=body)
+
+    plugin, _ = create_plugin(
+        WPushNotifier, {"apikey": "k"} if options is None else options
+    )
+    plugin.client_factory = lambda: httpx2.AsyncClient(  # type: ignore[method-assign]
+        transport=httpx2.MockTransport(handler)
+    )
+    return plugin.channel()(NotifierSettings())
+
+
+class TestChannel(NotifierContract):
+    def create(self):
+        return channel(200, {"code": 0, "message": "ok"})
+
+    def create_failing(self):
+        return channel(200, {"code": 1, "message": "apikey 无效"})
+
+
+def test_send_posts_payload_with_topic_code():
+    seen: list[httpx2.Request] = []
+    ch = channel(
+        200,
+        {"code": 0},
+        {"apikey": "k", "channel": "sms", "topic_code": "g1"},
+        seen,
+    )
+    msg = NotificationMessage(kind="event", title="标题", body="正文")
+    assert asyncio.run(ch.send(msg)) is True
+    [request] = seen
+    assert str(request.url) == API_URL
+    assert json.loads(request.content) == {
+        "apikey": "k",
+        "title": "标题",
+        "content": "正文",
+        "channel": "sms",
+        "topic_code": "g1",
+    }
+
+
+def test_test_reports_api_message_on_http_error():
+    ok, detail = asyncio.run(channel(500, {"message": "内部错误"}).test())
+    assert ok is False
+    assert "内部错误" in detail
+
+
+def test_send_without_apikey_returns_false_without_request():
+    seen: list[httpx2.Request] = []
+    ch = channel(200, {"code": 0}, {}, seen)
+    assert asyncio.run(ch.test()) == (False, "未填写 apikey")
+    assert seen == []
+
+
+@pytest.mark.parametrize(
+    ("status", "body", "expected"),
+    [(502, "<html>Bad Gateway</html>", "HTTP 502"), (200, "<html>", "不是预期的 JSON")],
+)
+def test_test_non_json_response_reports_status(status, body, expected):
+    ok, detail = asyncio.run(channel(status, body).test())
+    assert ok is False
+    assert expected in detail
--- a/examples/plugins/wpush-notifier/wpush_notifier/init.py
+++ b/examples/plugins/wpush-notifier/wpush_notifier/init.py
@@ -0,0 +1,90 @@
+"""WPUSH 通知渠道。
+
+POST JSON 到 WPUSH 接口；HTTP 2xx 且返回 JSON 的 ``code == 0`` 才算成功。
+接口文档：https://wpush.cn/docs
+"""
+
+from collections.abc import Callable
+
+import httpx2
+from pydantic import BaseModel, Field
+
+from ab_sdk import Plugin, points, provider, secret_field
+from ab_sdk.notify import NotificationMessage, NotifierFactory, NotifierSettings
+
+API_URL = "https://api.wpush.cn/api/v1/send"
+
+
+class Options(BaseModel):
+    apikey: str = secret_field(title="API Key", description="WPUSH 控制台中的 apikey")
+    channel: str = Field(
+        "wechat", title="推送渠道", description="如 wechat、sms、mail，见 WPUSH 文档"
+    )
+    topic_code: str = Field(
+        "", title="群组编码", description="可选：推送到群组时填写 topic_code"
+    )
+
+
+class Channel:
+    def __init__(
+        self,
+        options: Options,
+        settings: NotifierSettings,
+        client_factory: Callable[[], httpx2.AsyncClient],
+    ) -> None:
+        self.options = options
+        self.settings = settings
+        self.client_factory = client_factory
+
+    async def _post(self, message: NotificationMessage) -> tuple[bool, str]:
+        opts = self.options
+        if not opts.apikey:
+            return False, "未填写 apikey"
+        payload = {
+            "apikey": opts.apikey,
+            "title": message.title,
+            "content": message.body,
+            "channel": opts.channel.strip() or "wechat",
+        }
+        if opts.topic_code.strip():
+            payload["topic_code"] = opts.topic_code.strip()
+        try:
+            async with self.client_factory() as client:
+                resp = await client.post(API_URL, json=payload)
+        except httpx2.HTTPError as e:
+            return False, f"请求失败：{e}"
+        try:
+            body = resp.json()
+        except ValueError:
+            body = None
+        if resp.is_success and isinstance(body, dict) and body.get("code") == 0:
+            return True, "ok"
+        detail = body.get("message") if isinstance(body, dict) else None
+        if not resp.is_success:
+            # 网关的错误页不是 JSON，先报 HTTP 状态
+            detail = f"HTTP {resp.status_code}" + (f"，{detail}" if detail else "")
+        return False, f"WPUSH 返回错误：{detail or '响应不是预期的 JSON'}"
+
+    async def send(self, message: NotificationMessage) -> bool:
+        ok, _ = await self._post(message)
+        return ok
+
+    async def test(self) -> tuple[bool, str]:
+        return await self._post(
+            NotificationMessage(
+                kind="event", title="AutoBangumi 通知测试", body="通知测试成功！"
+            )
+        )
+
+
+class WPushNotifier(Plugin[Options]):
+    config_model = Options
+
+    # 测试可替换为带 MockTransport 的客户端
+    client_factory: Callable[[], httpx2.AsyncClient] = staticmethod(
+        lambda: httpx2.AsyncClient(timeout=10)
+    )
+
+    @provider(points.NOTIFIER, id="wpush")
+    def channel(self) -> NotifierFactory:
+        return lambda settings: Channel(self.config, settings, self.client_factory)
--- a/webui/src/api/notification.ts
+++ b/webui/src/api/notification.ts
@@ -17,6 +17,10 @@ export interface TestProviderConfigRequest {
  api_token?: string;
  template?: string;
  url?: string;
+  sound?: string;
+  group?: string;
+  level?: string;
+  icon?: string;
 }

 export interface TestResponse {
--- a/webui/src/components/setting/tests/config-notification.test.ts
+++ b/webui/src/components/setting/tests/config-notification.test.ts
@@ -1,4 +1,4 @@
-import { flushPromises, mount } from '@vue/test-utils';
+import { type VueWrapper, flushPromises, mount } from '@vue/test-utils';
 import { ref } from 'vue';
 import ConfigNotification from '../config-notification.vue';
 import { apiNotification } from '@/api/notification';
@@ -52,6 +52,7 @@ function mountPage() {
        },
        NSelect: {
          props: ['value'],
+          emits: ['update:value'],
          template: '<select class="type" />',
        },
      },
@@ -122,6 +123,24 @@ describe('config-notification', () => {
    expect(w.find('.test-result').exists()).toBe(false);
  });

+  // Bark 也读 url（点击跳转）：切换类型后不能带着上一类型的 Webhook URL
+  it('should reset type-specific fields when the type changes in the add dialog', async () => {
+    const w = mountPage();
+    const add = w
+      .findAll('button')
+      .find((b) => b.text() === 'config.notification_set.add_provider')!;
+    await add.trigger('click');
+    const vm = w.vm as any;
+    vm.newProvider.type = 'webhook';
+    vm.newProvider.url = 'https://hook.example.com';
+    vm.newProvider.template = '{"t": 1}';
+    await flushPromises();
+    const select = w.findComponent('.provider-type-select') as VueWrapper;
+    select.vm.$emit('update:value', 'bark');
+    await flushPromises();
+    expect(vm.newProvider).toEqual({ type: 'bark', enabled: true });
+  });
+
  it('should show a credentials hint and no Test button for plugin types', async () => {
    const w = mountPage();
    const add = w
--- a/webui/src/components/setting/config-notification.vue
+++ b/webui/src/components/setting/config-notification.vue
@@ -56,8 +56,10 @@ const providerFields: Record<
  string,
  {
    key: keyof NotificationProviderConfig;
-    label: string;
+    label: string | (() => string);
    placeholder: string;
+    // 有选项时渲染为可清空的下拉框
+    options?: string[];
  }[]
 > = {
  telegram: [
@@ -78,6 +80,32 @@ const providerFields: Record<
      label: 'Server URL (optional)',
      placeholder: 'https://api.day.app',
    },
+    {
+      key: 'sound',
+      label: () => t('config.notification_set.bark_sound'),
+      placeholder: 'minuet',
+    },
+    {
+      key: 'group',
+      label: () => t('config.notification_set.bark_group'),
+      placeholder: 'AutoBangumi',
+    },
+    {
+      key: 'level',
+      label: () => t('config.notification_set.bark_level'),
+      placeholder: 'active',
+      options: ['active', 'timeSensitive', 'passive', 'critical'],
+    },
+    {
+      key: 'icon',
+      label: () => t('config.notification_set.bark_icon'),
+      placeholder: 'https://example.com/icon.png',
+    },
+    {
+      key: 'url',
+      label: () => t('config.notification_set.bark_url'),
+      placeholder: 'https://example.com',
+    },
  ],
  'server-chan': [{ key: 'token', label: 'SendKey', placeholder: 'sendkey' }],
  wecom: [
@@ -182,6 +210,12 @@ function openAddDialog() {
  showAddDialog.value = true;
 }

+// 新增时切换类型：清掉上一类型的字段。Bark 也读 url（点击跳转），
+// 残留的 Webhook URL 会被一起发出去
+function changeType(type: NotificationProviderConfig['type']) {
+  newProvider.value = { type, enabled: newProvider.value.enabled };
+}
+
 function openEditDialog(index: number) {
  editingIndex.value = index;
  newProvider.value = { ...providers.value[index] };
@@ -382,9 +416,10 @@ function getFieldsForType(type: string) {
      <div class="space-y-16">
        <ab-field :label="$t('config.notification_set.type')">
          <NSelect
-            v-model:value="newProvider.type"
+            :value="newProvider.type"
            :options="providerTypes"
            class="provider-type-select"
+            @update:value="changeType"
          />
        </ab-field>

@@ -393,8 +428,15 @@ function getFieldsForType(type: string) {
          :key="field.key"
          :label="field.label"
        >
+          <NSelect
+            v-if="field.options"
+            v-model:value="(newProvider as any)[field.key]"
+            :options="field.options.map((value) => ({ label: value, value }))"
+            :placeholder="field.placeholder"
+            clearable
+          />
          <ab-input
-            v-if="field.key !== 'template'"
+            v-else-if="field.key !== 'template'"
            v-model="(newProvider as any)[field.key]"
            :placeholder="field.placeholder"
          />
@@ -462,8 +504,15 @@ function getFieldsForType(type: string) {
          :key="field.key"
          :label="field.label"
        >
+          <NSelect
+            v-if="field.options"
+            v-model:value="(newProvider as any)[field.key]"
+            :options="field.options.map((value) => ({ label: value, value }))"
+            :placeholder="field.placeholder"
+            clearable
+          />
          <ab-input
-            v-if="field.key !== 'template'"
+            v-else-if="field.key !== 'template'"
            v-model="(newProvider as any)[field.key]"
            :placeholder="field.placeholder"
          />
--- a/webui/src/i18n/en.json
+++ b/webui/src/i18n/en.json
@@ -119,7 +119,12 @@
      "test_success": "Test successful",
      "test_failed": "Test failed",
      "plugin_label": "{id} (plugin)",
-      "plugin_hint": "This channel comes from a plugin and has no fields here. Enter its credentials in the plugin's options in the Plugins section."
+      "plugin_hint": "This channel comes from a plugin and has no fields here. Enter its credentials in the plugin's options in the Plugins section.",
+      "bark_sound": "Sound (optional)",
+      "bark_group": "Group (optional)",
+      "bark_level": "Interruption level (optional)",
+      "bark_icon": "Icon URL (optional, defaults to the poster)",
+      "bark_url": "URL to open on tap (optional)"
    },
    "parser_set": {
      "enable": "Enable",
@@ -245,7 +250,8 @@
      "permission_labels": {
        "fs_write": "Write files",
        "network": "Network access",
-        "filesystem": "File system access"
+        "filesystem": "File system access",
+        "subprocess": "Run external commands"
      },
      "disable": "Disable",
      "disable_confirm_title": "Disable {name}?",
--- a/webui/src/i18n/zh-CN.json
+++ b/webui/src/i18n/zh-CN.json
@@ -119,7 +119,12 @@
      "test_success": "测试成功",
      "test_failed": "测试失败",
      "plugin_label": "{id}（插件）",
-      "plugin_hint": "此渠道由插件提供，这里没有可填字段。请在插件设置中该插件的配置里填写凭据。"
+      "plugin_hint": "此渠道由插件提供，这里没有可填字段。请在插件设置中该插件的配置里填写凭据。",
+      "bark_sound": "铃声（可选）",
+      "bark_group": "分组（可选）",
+      "bark_level": "中断级别（可选）",
+      "bark_icon": "图标地址（可选，默认用海报）",
+      "bark_url": "点击打开的地址（可选）"
    },
    "parser_set": {
      "enable": "启用",
@@ -245,7 +250,8 @@
      "permission_labels": {
        "fs_write": "写入文件",
        "network": "访问网络",
-        "filesystem": "访问文件系统"
+        "filesystem": "访问文件系统",
+        "subprocess": "运行外部命令"
      },
      "disable": "停用",
      "disable_confirm_title": "停用 {name}？",
--- a/webui/types/config.ts
+++ b/webui/types/config.ts
@@ -95,6 +95,11 @@ export interface NotificationProviderConfig {
  api_token?: string;
  template?: string;
  url?: string;
+  // Bark 的可选推送参数
+  sound?: string;
+  group?: string;
+  level?: string;
+  icon?: string;
 }

 export interface Notification {