feat(plugins): 新增 WPUSH、OneBot、Apprise 官方通知插件示例

- wpush-notifier：POST JSON 到 WPUSH 接口，code == 0 才算成功
- onebot-notifier：OneBot v11 私聊 / 群消息，可附带 poster_url 海报
- apprise-notifier：调用 Apprise API 服务器，支持 URL 列表与配置键
- 宿主加载测试与示例文档登记三个新渠道

backend/src/test/test_plugin_examples.py

@@ -117,7 +117,12 @@ 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",
+        }
         assert set(registry.providers(points.SEARCH_SITE)) == {"my-tracker"}
         assert [h.plugin_id for h in registry.hooks(points.HTTP_REQUEST)] == [
             "custom-rss-site"

docs/dev/plugins/examples.md

@@ -9,6 +9,9 @@
 | `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 列表与配置键两种方式 |
 | `manual-pick` | 前端挂载点 `bangumi.detail.tab`、插件路由、插件 KV、事件 |
 
 ## 安装一个示例

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`；填写后忽略配置键 |
+| `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) 中的实现

examples/plugins/apprise-notifier/apprise_notifier/__init__.py

@@ -0,0 +1,91 @@
+"""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
+from ab_sdk.notify import NotificationMessage, NotifierFactory, NotifierSettings
+
+
+class Options(BaseModel):
+    server: str = Field(
+        "http://apprise:8000", title="服务器", description="Apprise API 服务器地址"
+    )
+    urls: list[str] = Field(
+        default_factory=list,
+        title="Apprise URL",
+        description="如 tgram://bottoken/ChatID；填写后忽略配置键",
+    )
+    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}
+        if opts.urls:
+            url = f"{server}/notify/"
+            # Apprise 接受逗号分隔的 URL 字符串
+            payload["urls"] = ", ".join(opts.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)

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"]

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"]

examples/plugins/apprise-notifier/tests/test_apprise_notifier.py

@@ -0,0 +1,63 @@
+import asyncio
+import json
+
+import httpx2
+import pytest
+from apprise_notifier import AppriseNotifier
+
+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", "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 == []

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 实现能访问这个地址；它会自行下载图片。未配置时只发送文字。
+
+## 安装
+
+```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) 中的实现

examples/plugins/onebot-notifier/onebot_notifier/__init__.py

@@ -0,0 +1,113 @@
+"""OneBot v11 通知渠道。
+
+POST JSON 到 ``{url}/send_private_msg`` 或 ``{url}/send_group_msg``，消息为消息段
+数组：文字，以及（开启海报且宿主给出海报地址时）一张图片。HTTP 2xx 且返回
+``status == "ok"`` 或 ``retcode == 0`` 才算成功。
+协议文档：https://github.com/botuniverse/onebot-11
+"""
+
+from collections.abc import Callable
+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
+
+    def _segments(self, message: NotificationMessage) -> list[dict[str, Any]]:
+        # 单集通知的正文已含番剧名；系统事件的标题与正文分开，拼在一起发
+        text = (
+            message.body
+            if message.kind == "episode"
+            else f"{message.title}\n{message.body}"
+        )
+        segments: list[dict[str, Any]] = []
+        # SDK 只给出公开海报 URL（没有本地路径），由 OneBot 实现自行下载
+        if self.options.send_poster and message.poster_url:
+            segments.append({"type": "image", "data": {"file": message.poster_url}})
+        segments.append({"type": "text", "data": {"text": text}})
+        return segments
+
+    async def _post(self, message: NotificationMessage) -> tuple[bool, str]:
+        opts = self.options
+        if not opts.url or not opts.target:
+            return False, "未填写 API 地址或目标"
+        id_key = "group_id" if opts.message_type == "group" else "user_id"
+        headers = {}
+        if opts.access_token:
+            headers["Authorization"] = f"Bearer {opts.access_token}"
+        try:
+            async with self.client_factory() as client:
+                resp = await client.post(
+                    f"{opts.url.rstrip('/')}/send_{opts.message_type}_msg",
+                    json={id_key: opts.target, "message": self._segments(message)},
+                    headers=headers,
+                )
+            body = resp.json()
+        except (httpx2.HTTPError, ValueError) as e:
+            return False, f"请求失败：{e}"
+        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
+        )
+        return False, f"OneBot 返回错误：{detail or resp.status_code}"
+
+    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)

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"]

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"]

examples/plugins/onebot-notifier/tests/test_onebot_notifier.py

@@ -0,0 +1,93 @@
+import asyncio
+import json
+
+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, options: dict | None = None, seen=None):
+    def handler(request: httpx2.Request) -> httpx2.Response:
+        if seen is not None:
+            seen.append(request)
+        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集"}}],
+    }
+
+
+@pytest.mark.parametrize(
+    ("send_poster", "poster_url", "with_image"),
+    [(True, POSTER, True), (False, POSTER, False), (True, None, False)],
+)
+def test_send_adds_image_segment_only_with_poster(send_poster, poster_url, with_image):
+    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
+    segments = json.loads(seen[0].content)["message"]
+    image = {"type": "image", "data": {"file": POSTER}}
+    assert (segments[0] == image) is with_image
+    assert segments[-1]["type"] == "text"
+
+
+@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 == []

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) 中的实现

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"]

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"]

examples/plugins/wpush-notifier/tests/test_wpush_notifier.py

@@ -0,0 +1,65 @@
+import asyncio
+import json
+
+import httpx2
+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, options: dict | None = None, seen=None):
+    def handler(request: httpx2.Request) -> httpx2.Response:
+        if seen is not None:
+            seen.append(request)
+        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 == []

examples/plugins/wpush-notifier/wpush_notifier/__init__.py

@@ -0,0 +1,84 @@
+"""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)
+            body = resp.json()
+        except (httpx2.HTTPError, ValueError) as e:
+            return False, f"请求失败：{e}"
+        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
+        return False, f"WPUSH 返回错误：{detail or resp.status_code}"
+
+    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)