From 403cf94572819849ef89c0ffdd71dbdde544af5a Mon Sep 17 00:00:00 2001 From: Estrella Pan Date: Wed, 21 Oct 2026 08:53:28 +0200 Subject: [PATCH] =?UTF-8?q?feat(plugins):=20=E6=96=B0=E5=A2=9E=20WPUSH?= =?UTF-8?q?=E3=80=81OneBot=E3=80=81Apprise=20=E5=AE=98=E6=96=B9=E9=80=9A?= =?UTF-8?q?=E7=9F=A5=E6=8F=92=E4=BB=B6=E7=A4=BA=E4=BE=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 | 7 +- docs/dev/plugins/examples.md | 3 + examples/plugins/apprise-notifier/README.md | 45 +++++++ .../apprise_notifier/__init__.py | 91 ++++++++++++++ examples/plugins/apprise-notifier/plugin.toml | 10 ++ .../plugins/apprise-notifier/pyproject.toml | 13 ++ .../tests/test_apprise_notifier.py | 63 ++++++++++ examples/plugins/onebot-notifier/README.md | 40 +++++++ .../onebot_notifier/__init__.py | 113 ++++++++++++++++++ examples/plugins/onebot-notifier/plugin.toml | 10 ++ .../plugins/onebot-notifier/pyproject.toml | 13 ++ .../tests/test_onebot_notifier.py | 93 ++++++++++++++ examples/plugins/wpush-notifier/README.md | 33 +++++ examples/plugins/wpush-notifier/plugin.toml | 10 ++ .../plugins/wpush-notifier/pyproject.toml | 13 ++ .../tests/test_wpush_notifier.py | 65 ++++++++++ .../wpush-notifier/wpush_notifier/__init__.py | 84 +++++++++++++ 17 files changed, 705 insertions(+), 1 deletion(-) create mode 100644 examples/plugins/apprise-notifier/README.md create mode 100644 examples/plugins/apprise-notifier/apprise_notifier/__init__.py create mode 100644 examples/plugins/apprise-notifier/plugin.toml create mode 100644 examples/plugins/apprise-notifier/pyproject.toml create mode 100644 examples/plugins/apprise-notifier/tests/test_apprise_notifier.py create mode 100644 examples/plugins/onebot-notifier/README.md create mode 100644 examples/plugins/onebot-notifier/onebot_notifier/__init__.py create mode 100644 examples/plugins/onebot-notifier/plugin.toml create mode 100644 examples/plugins/onebot-notifier/pyproject.toml create mode 100644 examples/plugins/onebot-notifier/tests/test_onebot_notifier.py create mode 100644 examples/plugins/wpush-notifier/README.md create mode 100644 examples/plugins/wpush-notifier/plugin.toml create mode 100644 examples/plugins/wpush-notifier/pyproject.toml create mode 100644 examples/plugins/wpush-notifier/tests/test_wpush_notifier.py create mode 100644 examples/plugins/wpush-notifier/wpush_notifier/__init__.py diff --git a/backend/src/test/test_plugin_examples.py b/backend/src/test/test_plugin_examples.py index 5c78b17f..2c283f0f 100644 --- a/backend/src/test/test_plugin_examples.py +++ b/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" diff --git a/docs/dev/plugins/examples.md b/docs/dev/plugins/examples.md index 95860a1e..c4ef5847 100644 --- a/docs/dev/plugins/examples.md +++ b/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、事件 | ## 安装一个示例 diff --git a/examples/plugins/apprise-notifier/README.md b/examples/plugins/apprise-notifier/README.md new file mode 100644 index 00000000..2077cea2 --- /dev/null +++ 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`；填写后忽略配置键 | +| `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) 中的实现 diff --git a/examples/plugins/apprise-notifier/apprise_notifier/__init__.py b/examples/plugins/apprise-notifier/apprise_notifier/__init__.py new file mode 100644 index 00000000..9b5e6e20 --- /dev/null +++ b/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) diff --git a/examples/plugins/apprise-notifier/plugin.toml b/examples/plugins/apprise-notifier/plugin.toml new file mode 100644 index 00000000..9bf860ec --- /dev/null +++ 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"] diff --git a/examples/plugins/apprise-notifier/pyproject.toml b/examples/plugins/apprise-notifier/pyproject.toml new file mode 100644 index 00000000..74ea2cee --- /dev/null +++ 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"] diff --git a/examples/plugins/apprise-notifier/tests/test_apprise_notifier.py b/examples/plugins/apprise-notifier/tests/test_apprise_notifier.py new file mode 100644 index 00000000..1d2057d1 --- /dev/null +++ b/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 == [] diff --git a/examples/plugins/onebot-notifier/README.md b/examples/plugins/onebot-notifier/README.md new file mode 100644 index 00000000..1dfb2844 --- /dev/null +++ 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 ` +- 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) 中的实现 diff --git a/examples/plugins/onebot-notifier/onebot_notifier/__init__.py b/examples/plugins/onebot-notifier/onebot_notifier/__init__.py new file mode 100644 index 00000000..c93b8919 --- /dev/null +++ b/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) diff --git a/examples/plugins/onebot-notifier/plugin.toml b/examples/plugins/onebot-notifier/plugin.toml new file mode 100644 index 00000000..7fb32e56 --- /dev/null +++ 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"] diff --git a/examples/plugins/onebot-notifier/pyproject.toml b/examples/plugins/onebot-notifier/pyproject.toml new file mode 100644 index 00000000..f02ae942 --- /dev/null +++ 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"] diff --git a/examples/plugins/onebot-notifier/tests/test_onebot_notifier.py b/examples/plugins/onebot-notifier/tests/test_onebot_notifier.py new file mode 100644 index 00000000..a48a54ff --- /dev/null +++ b/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 == [] diff --git a/examples/plugins/wpush-notifier/README.md b/examples/plugins/wpush-notifier/README.md new file mode 100644 index 00000000..a6ba964f --- /dev/null +++ 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) 中的实现 diff --git a/examples/plugins/wpush-notifier/plugin.toml b/examples/plugins/wpush-notifier/plugin.toml new file mode 100644 index 00000000..b371d3f8 --- /dev/null +++ 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"] diff --git a/examples/plugins/wpush-notifier/pyproject.toml b/examples/plugins/wpush-notifier/pyproject.toml new file mode 100644 index 00000000..f2b6fef0 --- /dev/null +++ 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"] diff --git a/examples/plugins/wpush-notifier/tests/test_wpush_notifier.py b/examples/plugins/wpush-notifier/tests/test_wpush_notifier.py new file mode 100644 index 00000000..aecff6d8 --- /dev/null +++ b/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 == [] diff --git a/examples/plugins/wpush-notifier/wpush_notifier/__init__.py b/examples/plugins/wpush-notifier/wpush_notifier/__init__.py new file mode 100644 index 00000000..e2a3f55a --- /dev/null +++ b/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)