From a1adf237b8afa7dcb5234defcb5e9a6042b8ba62 Mon Sep 17 00:00:00 2001 From: Estrella Pan Date: Thu, 8 Oct 2026 14:50:59 +0200 Subject: [PATCH] =?UTF-8?q?feat(4.0-P7):=20=E6=96=B0=E5=A2=9E=205=20?= =?UTF-8?q?=E4=B8=AA=E7=A4=BA=E4=BE=8B=E6=8F=92=E4=BB=B6=EF=BC=8C=E9=80=90?= =?UTF-8?q?=E4=B8=AA=E8=BF=90=E8=A1=8C=E5=A5=91=E7=BA=A6=E4=B8=8E=E8=A1=8C?= =?UTF-8?q?=E4=B8=BA=E6=B5=8B=E8=AF=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - webhook-on-event、custom-rss-site、template-rename、nfo-writer、ntfy-notifier - scripts/test_example_plugins.sh 对每个示例运行 ab-plugin validate 与自带测试 - 宿主侧测试把全部示例载入 PluginManager，断言 Provider 与钩子已登记 - NotifierContract 发现 ntfy 示例的中文标题请求头缺陷，已改用 bytes 请求头 --- backend/src/test/test_plugin_examples.py | 32 +++++- examples/plugins/custom-rss-site/README.md | 21 ++++ .../custom_rss_site/__init__.py | 41 +++++++ examples/plugins/custom-rss-site/plugin.toml | 10 ++ .../plugins/custom-rss-site/pyproject.toml | 13 +++ .../tests/test_custom_rss_site.py | 30 +++++ examples/plugins/nfo-writer/README.md | 23 ++++ .../plugins/nfo-writer/nfo_writer/__init__.py | 82 ++++++++++++++ examples/plugins/nfo-writer/plugin.toml | 9 ++ examples/plugins/nfo-writer/pyproject.toml | 13 +++ .../nfo-writer/tests/test_nfo_writer.py | 48 ++++++++ examples/plugins/ntfy-notifier/README.md | 22 ++++ .../ntfy-notifier/ntfy_notifier/__init__.py | 69 ++++++++++++ examples/plugins/ntfy-notifier/plugin.toml | 10 ++ examples/plugins/ntfy-notifier/pyproject.toml | 13 +++ .../ntfy-notifier/tests/test_ntfy_notifier.py | 21 ++++ examples/plugins/template-rename/README.md | 22 ++++ examples/plugins/template-rename/plugin.toml | 9 ++ .../plugins/template-rename/pyproject.toml | 13 +++ .../template_rename/__init__.py | 105 ++++++++++++++++++ .../tests/test_template_rename.py | 58 ++++++++++ examples/plugins/webhook-on-event/README.md | 21 ++++ examples/plugins/webhook-on-event/plugin.toml | 9 ++ .../plugins/webhook-on-event/pyproject.toml | 13 +++ .../tests/test_webhook_on_event.py | 68 ++++++++++++ .../webhook_on_event/__init__.py | 68 ++++++++++++ scripts/test_example_plugins.sh | 14 +++ 27 files changed, 856 insertions(+), 1 deletion(-) create mode 100644 examples/plugins/custom-rss-site/README.md create mode 100644 examples/plugins/custom-rss-site/custom_rss_site/__init__.py create mode 100644 examples/plugins/custom-rss-site/plugin.toml create mode 100644 examples/plugins/custom-rss-site/pyproject.toml create mode 100644 examples/plugins/custom-rss-site/tests/test_custom_rss_site.py create mode 100644 examples/plugins/nfo-writer/README.md create mode 100644 examples/plugins/nfo-writer/nfo_writer/__init__.py create mode 100644 examples/plugins/nfo-writer/plugin.toml create mode 100644 examples/plugins/nfo-writer/pyproject.toml create mode 100644 examples/plugins/nfo-writer/tests/test_nfo_writer.py create mode 100644 examples/plugins/ntfy-notifier/README.md create mode 100644 examples/plugins/ntfy-notifier/ntfy_notifier/__init__.py create mode 100644 examples/plugins/ntfy-notifier/plugin.toml create mode 100644 examples/plugins/ntfy-notifier/pyproject.toml create mode 100644 examples/plugins/ntfy-notifier/tests/test_ntfy_notifier.py create mode 100644 examples/plugins/template-rename/README.md create mode 100644 examples/plugins/template-rename/plugin.toml create mode 100644 examples/plugins/template-rename/pyproject.toml create mode 100644 examples/plugins/template-rename/template_rename/__init__.py create mode 100644 examples/plugins/template-rename/tests/test_template_rename.py create mode 100644 examples/plugins/webhook-on-event/README.md create mode 100644 examples/plugins/webhook-on-event/plugin.toml create mode 100644 examples/plugins/webhook-on-event/pyproject.toml create mode 100644 examples/plugins/webhook-on-event/tests/test_webhook_on_event.py create mode 100644 examples/plugins/webhook-on-event/webhook_on_event/__init__.py create mode 100755 scripts/test_example_plugins.sh diff --git a/backend/src/test/test_plugin_examples.py b/backend/src/test/test_plugin_examples.py index 85afc608..5c78b17f 100644 --- a/backend/src/test/test_plugin_examples.py +++ b/backend/src/test/test_plugin_examples.py @@ -65,7 +65,7 @@ async def manager(tmp_path): async def test_manual_pick_routes_store_pick_and_publish_event(manager): from module.core.plugin_routes import PluginRoutes - [status] = manager.statuses() + [status] = [s for s in manager.statuses() if s.id == "manual-pick"] assert status.state == "active" assert [(pid, ui.slot) for pid, ui in manager.ui_slots()] == [ ("manual-pick", "bangumi.detail.tab") @@ -94,3 +94,33 @@ async def test_manual_pick_routes_store_pick_and_publish_event(manager): assert client.get(base).json() == {"torrent_id": 9} await manager.bus.drain() assert [(e.bangumi_id, e.torrent_id) for e in seen] == [(3, 9)] + + +async def test_all_examples_load_in_host_with_expected_providers(tmp_path): + """每个示例都能在真实宿主里加载并登记预期的扩展（SDK 侧的契约套件在 + ``scripts/test_example_plugins.sh`` 里跑）。""" + from module.plugin.host import POINTS + + ids = sorted(p.name for p in EXAMPLES.iterdir() if (p / "plugin.toml").is_file()) + registry = ExtensionRegistry() + for point in POINTS: + registry.declare(point) + mgr = PluginManager( + SimpleNamespace( + plugins=Plugins(allow_unsigned=True, enabled={i: True for i in ids}) + ), + registry=registry, + discover_fn=lambda: (_discover(EXAMPLES, tmp_path / "none"), []), + data_root=tmp_path, + ) + await mgr.start() + 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.SEARCH_SITE)) == {"my-tracker"} + assert [h.plugin_id for h in registry.hooks(points.HTTP_REQUEST)] == [ + "custom-rss-site" + ] + finally: + await mgr.stop() diff --git a/examples/plugins/custom-rss-site/README.md b/examples/plugins/custom-rss-site/README.md new file mode 100644 index 00000000..7bb3922b --- /dev/null +++ b/examples/plugins/custom-rss-site/README.md @@ -0,0 +1,21 @@ +# 自定义 RSS 站点（示例插件） + +给搜索框加一个站点，并为该站点的请求附带 Cookie。演示： + +- `@provider(points.SEARCH_SITE)`：返回 `SearchSite` +- `@hook(points.HTTP_REQUEST)`：transform 钩子，返回修改后的请求副本；返回 `None` 表示不修改 +- 契约套件 `SearchSiteContract` + +## 安装 + +```bash +cp -r examples/plugins/custom-rss-site config/plugins/local/ +``` + +在「设置 → 插件」中开启「允许未签名插件」，启用并填写搜索地址与 Cookie。 + +## 测试 + +```bash +cd examples/plugins/custom-rss-site && uv run pytest +``` diff --git a/examples/plugins/custom-rss-site/custom_rss_site/__init__.py b/examples/plugins/custom-rss-site/custom_rss_site/__init__.py new file mode 100644 index 00000000..7396892f --- /dev/null +++ b/examples/plugins/custom-rss-site/custom_rss_site/__init__.py @@ -0,0 +1,41 @@ +"""示例插件：自定义 RSS 站点。 + +- ``search_site``：搜索框多出站点 ``my-tracker``。``%s`` 被替换为关键词。 +- ``http.request``：宿主向该站点发 GET 请求（RSS、种子文件）时附带 Cookie， + 用于需要登录的私有站。只改本站点的请求，其它请求原样放行。 +""" + +from dataclasses import replace +from urllib.parse import urlparse + +from pydantic import BaseModel, Field + +from ab_sdk import Plugin, hook, points, provider, secret_field +from ab_sdk.ingest import HttpRequest +from ab_sdk.search import SearchSite + + +class Options(BaseModel): + url: str = Field( + "https://tracker.example.org/rss?search=%s", + title="搜索地址", + description="RSS 搜索地址，%s 代表关键词", + ) + parser: str = Field("tmdb", title="元数据源", pattern="^(mikan|tmdb)$") + cookie: str = secret_field(title="Cookie", description="私有站登录后的 Cookie") + + +class CustomRssSite(Plugin[Options]): + config_model = Options + + @provider(points.SEARCH_SITE, id="my-tracker") + def site(self) -> SearchSite: + return SearchSite(url=self.config.url, parser=self.config.parser) + + @hook(points.HTTP_REQUEST) + def add_cookie(self, request: HttpRequest) -> HttpRequest | None: + cfg = self.config + host = urlparse(cfg.url.replace("%s", "x")).hostname + if not cfg.cookie or urlparse(request.url).hostname != host: + return None # None 表示不修改 + return replace(request, headers={**request.headers, "Cookie": cfg.cookie}) diff --git a/examples/plugins/custom-rss-site/plugin.toml b/examples/plugins/custom-rss-site/plugin.toml new file mode 100644 index 00000000..987c6029 --- /dev/null +++ b/examples/plugins/custom-rss-site/plugin.toml @@ -0,0 +1,10 @@ +[plugin] +id = "custom-rss-site" +name = "自定义 RSS 站点" +version = "0.1.0" +sdk = ">=0.5,<1" +entry = "custom_rss_site:CustomRssSite" +description = "示例插件：给搜索框加一个站点，并为该站点的请求附带 Cookie。演示 search_site 与 http.request" +authors = ["AutoBangumi"] +permissions = ["network"] +extension_points = ["search_site", "http.request"] diff --git a/examples/plugins/custom-rss-site/pyproject.toml b/examples/plugins/custom-rss-site/pyproject.toml new file mode 100644 index 00000000..fb5453e9 --- /dev/null +++ b/examples/plugins/custom-rss-site/pyproject.toml @@ -0,0 +1,13 @@ +# 只用于本地开发与测试，不会被打进插件包（ab-plugin pack 排除它） +[project] +name = "ab-plugin-custom-rss-site" +version = "0.1.0" +requires-python = ">=3.13" +dependencies = ["autobangumi-sdk"] + +[dependency-groups] +dev = ["pytest>=8.0.0"] + +[tool.pytest.ini_options] +pythonpath = ["."] +testpaths = ["tests"] diff --git a/examples/plugins/custom-rss-site/tests/test_custom_rss_site.py b/examples/plugins/custom-rss-site/tests/test_custom_rss_site.py new file mode 100644 index 00000000..e99b9534 --- /dev/null +++ b/examples/plugins/custom-rss-site/tests/test_custom_rss_site.py @@ -0,0 +1,30 @@ +import pytest +from custom_rss_site import CustomRssSite + +from ab_sdk.ingest import HttpRequest +from ab_sdk.testing import SearchSiteContract, create_plugin + + +class TestSite(SearchSiteContract): + def create(self): + plugin, _ = create_plugin(CustomRssSite) + return plugin.site() + + +@pytest.mark.parametrize( + ("url", "cookie"), + [ + ("https://tracker.example.org/rss?search=a", "uid=1"), + ("https://other.example.com/feed", None), + ], +) +def test_add_cookie_only_for_own_host(url, cookie): + plugin, _ = create_plugin(CustomRssSite, {"cookie": "uid=1"}) + result = plugin.add_cookie(HttpRequest("GET", url, {})) + assert (result.headers["Cookie"] if result else None) == cookie + + +def test_add_cookie_empty_cookie_leaves_request_alone(): + plugin, _ = create_plugin(CustomRssSite) + request = HttpRequest("GET", "https://tracker.example.org/rss", {}) + assert plugin.add_cookie(request) is None diff --git a/examples/plugins/nfo-writer/README.md b/examples/plugins/nfo-writer/README.md new file mode 100644 index 00000000..d9103875 --- /dev/null +++ b/examples/plugins/nfo-writer/README.md @@ -0,0 +1,23 @@ +# NFO 写入（示例插件） + +种子整理完成后，在每个正片旁写一个 Kodi / Jellyfin 格式的 `.nfo`（只含剧名、季、集）。演示： + +- `@subscribe(TorrentOrganized.kind)`：事件为「至少一次」投递，插件自己保证幂等（`.nfo` 已存在就跳过） +- 事件里的路径是下载器视角，用 `path_from` / `path_to` 换成本地路径 +- 文件操作放进 `asyncio.to_thread`，不阻塞事件循环 + +刷新媒体库不需要写插件：内置插件 `media-server-refresh` 已经支持 Jellyfin / Emby / Plex。 + +## 安装 + +```bash +cp -r examples/plugins/nfo-writer config/plugins/local/ +``` + +在「设置 → 插件」中开启「允许未签名插件」并启用。 + +## 测试 + +```bash +cd examples/plugins/nfo-writer && uv run pytest +``` diff --git a/examples/plugins/nfo-writer/nfo_writer/__init__.py b/examples/plugins/nfo-writer/nfo_writer/__init__.py new file mode 100644 index 00000000..3e2de20b --- /dev/null +++ b/examples/plugins/nfo-writer/nfo_writer/__init__.py @@ -0,0 +1,82 @@ +"""示例插件：NFO 写入。 + +订阅 ``torrent.organized``。对事件里每个正片 ``<名>.mkv``，在旁边写 ``<名>.nfo`` +（Kodi / Jellyfin 的 ``episodedetails``）。季号、集数取自文件名中的 ``SxxEyy`` +（内置重命名方式 pn / advance / template 的默认输出都含它），剧名取自 +``Season N`` 的上一级目录。 + +- 事件里的路径是下载器视角：配置 ``path_from`` / ``path_to`` 换成 AutoBangumi + 看到的路径（两者都留空表示相同）。 +- 投递为「至少一次」：``.nfo`` 已存在就跳过，所以重复事件和用户手改的文件都安全。 +""" + +import asyncio +import re +from pathlib import Path +from xml.sax.saxutils import escape + +from pydantic import BaseModel, Field + +from ab_sdk import Plugin, subscribe +from ab_sdk.events import TorrentOrganized + +_EPISODE = re.compile(r"S(\d+)E(\d+(?:\.\d+)?)", re.IGNORECASE) + + +class Options(BaseModel): + path_from: str = Field( + "", title="下载器路径前缀", description="下载器看到的路径前缀；留空表示不映射" + ) + path_to: str = Field( + "", title="本地路径前缀", description="同一位置在 AutoBangumi 中的路径" + ) + + +def to_local(path: str, options: Options) -> Path: + src = options.path_from.rstrip("/") + if src and (path == src or path.startswith(src + "/")): + return Path(options.path_to.rstrip("/") + path[len(src) :]) + return Path(path) + + +def nfo_xml(media: Path) -> str | None: + """文件名里没有 SxxEyy 时返回 None（电影、特别篇等不写）。""" + match = _EPISODE.search(media.stem) + if match is None: + return None + show = media.parent.parent.name if media.parent.name.startswith("Season") else "" + return ( + '\n' + "\n" + f" {escape(show)}\n" + f" {int(match.group(1))}\n" + f" {match.group(2).lstrip('0') or '0'}\n" + "\n" + ) + + +def write_nfo(media: Path) -> bool: + """已存在或不适用时返回 False。``x`` 模式保证不覆盖已有文件。""" + content = nfo_xml(media) + nfo = media.with_suffix(".nfo") + if content is None or not media.parent.is_dir(): + return False + try: + with nfo.open("x", encoding="utf-8") as fh: + fh.write(content) + except FileExistsError: + return False + return True + + +class NfoWriter(Plugin[Options]): + config_model = Options + + @subscribe(TorrentOrganized.kind) + async def on_organized(self, event: TorrentOrganized) -> None: + for file in event.files: + if file.kind != "media": + continue + media = to_local(file.path, self.config) + if await asyncio.to_thread(write_nfo, media): + self.ctx.log.info("已写入 %s", media.with_suffix(".nfo")) diff --git a/examples/plugins/nfo-writer/plugin.toml b/examples/plugins/nfo-writer/plugin.toml new file mode 100644 index 00000000..7c896bbf --- /dev/null +++ b/examples/plugins/nfo-writer/plugin.toml @@ -0,0 +1,9 @@ +[plugin] +id = "nfo-writer" +name = "NFO 写入" +version = "0.1.0" +sdk = ">=0.5,<1" +entry = "nfo_writer:NfoWriter" +description = "示例插件：种子整理完成后，在每个正片旁写一个 Kodi / Jellyfin 格式的 .nfo。演示 torrent.organized 事件与路径映射" +authors = ["AutoBangumi"] +permissions = ["filesystem"] diff --git a/examples/plugins/nfo-writer/pyproject.toml b/examples/plugins/nfo-writer/pyproject.toml new file mode 100644 index 00000000..37bf364c --- /dev/null +++ b/examples/plugins/nfo-writer/pyproject.toml @@ -0,0 +1,13 @@ +# 只用于本地开发与测试，不会被打进插件包（ab-plugin pack 排除它） +[project] +name = "ab-plugin-nfo-writer" +version = "0.1.0" +requires-python = ">=3.13" +dependencies = ["autobangumi-sdk"] + +[dependency-groups] +dev = ["pytest>=8.0.0"] + +[tool.pytest.ini_options] +pythonpath = ["."] +testpaths = ["tests"] diff --git a/examples/plugins/nfo-writer/tests/test_nfo_writer.py b/examples/plugins/nfo-writer/tests/test_nfo_writer.py new file mode 100644 index 00000000..4694a827 --- /dev/null +++ b/examples/plugins/nfo-writer/tests/test_nfo_writer.py @@ -0,0 +1,48 @@ +import asyncio + +from nfo_writer import NfoWriter + +from ab_sdk.events import OrganizedFile, TorrentOrganized +from ab_sdk.testing import create_plugin + + +def organized(*files: OrganizedFile) -> TorrentOrganized: + return TorrentOrganized(torrent_hash="h", bangumi_id=1, files=files) + + +def run(plugin, event) -> None: + asyncio.run(plugin.on_organized(event)) + + +def test_on_organized_writes_nfo_next_to_media_with_path_map(tmp_path): + season = tmp_path / "Show & Co" / "Season 1" + season.mkdir(parents=True) + plugin, _ = create_plugin( + NfoWriter, {"path_from": "/downloads", "path_to": str(tmp_path)} + ) + event = organized( + OrganizedFile("/downloads/Show & Co/Season 1/Show S01E05.mkv", "media"), + OrganizedFile("/downloads/Show & Co/Season 1/Show S01E05.zh.ass", "subtitle"), + ) + run(plugin, event) + nfo = (season / "Show S01E05.nfo").read_text(encoding="utf-8") + assert "Show & Co" in nfo + assert "1" in nfo and "5" in nfo + assert [p.name for p in season.iterdir()] == ["Show S01E05.nfo"] + + +def test_on_organized_existing_nfo_is_not_overwritten(tmp_path): + media = tmp_path / "Show" / "Season 1" / "Show S01E01.mkv" + media.parent.mkdir(parents=True) + nfo = media.with_suffix(".nfo") + nfo.write_text("mine", encoding="utf-8") + plugin, _ = create_plugin(NfoWriter) + run(plugin, organized(OrganizedFile(str(media), "media"))) + assert nfo.read_text(encoding="utf-8") == "mine" + + +def test_on_organized_name_without_episode_marker_writes_nothing(tmp_path): + media = tmp_path / "Movie (2024).mkv" + plugin, _ = create_plugin(NfoWriter) + run(plugin, organized(OrganizedFile(str(media), "media"))) + assert list(tmp_path.iterdir()) == [] diff --git a/examples/plugins/ntfy-notifier/README.md b/examples/plugins/ntfy-notifier/README.md new file mode 100644 index 00000000..872be6c3 --- /dev/null +++ b/examples/plugins/ntfy-notifier/README.md @@ -0,0 +1,22 @@ +# ntfy 通知（示例插件） + +把通知发到 [ntfy](https://ntfy.sh) 主题。演示： + +- `@provider(points.NOTIFIER, id="ntfy")`：返回 `NotifierFactory` +- 凭据放在插件的 `config_model` 里，渠道条目只带通用模板 +- 后端拒绝时 `send` 返回 `False` 而不抛异常 +- 契约套件 `NotifierContract` + +## 安装 + +```bash +cp -r examples/plugins/ntfy-notifier config/plugins/local/ +``` + +在「设置 → 插件」中开启「允许未签名插件」，启用并填写主题；再到「设置 → 通知 → 添加渠道」选择类型 `ntfy`。 + +## 测试 + +```bash +cd examples/plugins/ntfy-notifier && uv run pytest +``` diff --git a/examples/plugins/ntfy-notifier/ntfy_notifier/__init__.py b/examples/plugins/ntfy-notifier/ntfy_notifier/__init__.py new file mode 100644 index 00000000..d2f2fdec --- /dev/null +++ b/examples/plugins/ntfy-notifier/ntfy_notifier/__init__.py @@ -0,0 +1,69 @@ +"""示例插件：ntfy 通知渠道。 + +渠道凭据放在插件自己的配置里（``server``、``topic``、``token``），渠道条目只带 +通用模板。后端拒绝消息时 ``send`` 返回 ``False``，不抛异常。 +""" + +from collections.abc import Callable + +import httpx +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("https://ntfy.sh", title="服务器") + topic: str = Field("", title="主题") + token: str = secret_field(title="访问令牌", description="私有主题的 Bearer 令牌") + + +class Channel: + def __init__( + self, + options: Options, + settings: NotifierSettings, + client_factory: Callable[[], httpx.AsyncClient], + ) -> None: + self.options = options + self.settings = settings + self.client_factory = client_factory + + async def send(self, message: NotificationMessage) -> bool: + opts = self.options + if not opts.topic: + return False + # 标题含中文：httpx 的 str 头值只接受 ASCII，用 bytes 发 UTF-8 + headers = {b"Title": message.title.encode("utf-8")} + if opts.token: + headers[b"Authorization"] = f"Bearer {opts.token}".encode() + try: + async with self.client_factory() as client: + resp = await client.post( + f"{opts.server.rstrip('/')}/{opts.topic}", + content=message.body.encode("utf-8"), + headers=headers, + ) + except httpx.HTTPError: + return False + return resp.is_success + + async def test(self) -> tuple[bool, str]: + ok = await self.send( + NotificationMessage(kind="event", title="AutoBangumi", body="通知测试") + ) + return ok, "ok" if ok else "发送失败" + + +class NtfyNotifier(Plugin[Options]): + config_model = Options + + # 测试可替换为带 MockTransport 的客户端 + client_factory: Callable[[], httpx.AsyncClient] = staticmethod( + lambda: httpx.AsyncClient(timeout=10) + ) + + @provider(points.NOTIFIER, id="ntfy") + def channel(self) -> NotifierFactory: + return lambda settings: Channel(self.config, settings, self.client_factory) diff --git a/examples/plugins/ntfy-notifier/plugin.toml b/examples/plugins/ntfy-notifier/plugin.toml new file mode 100644 index 00000000..f3aea74f --- /dev/null +++ b/examples/plugins/ntfy-notifier/plugin.toml @@ -0,0 +1,10 @@ +[plugin] +id = "ntfy-notifier" +name = "ntfy 通知" +version = "0.1.0" +sdk = ">=0.5,<1" +entry = "ntfy_notifier:NtfyNotifier" +description = "示例插件：把通知发到 ntfy 主题。在「设置 → 通知 → 添加渠道」选择类型 ntfy。演示 notifier 与 NotifierContract" +authors = ["AutoBangumi"] +permissions = ["network"] +extension_points = ["notifier"] diff --git a/examples/plugins/ntfy-notifier/pyproject.toml b/examples/plugins/ntfy-notifier/pyproject.toml new file mode 100644 index 00000000..1051d48e --- /dev/null +++ b/examples/plugins/ntfy-notifier/pyproject.toml @@ -0,0 +1,13 @@ +# 只用于本地开发与测试，不会被打进插件包（ab-plugin pack 排除它） +[project] +name = "ab-plugin-ntfy-notifier" +version = "0.1.0" +requires-python = ">=3.13" +dependencies = ["autobangumi-sdk"] + +[dependency-groups] +dev = ["pytest>=8.0.0"] + +[tool.pytest.ini_options] +pythonpath = ["."] +testpaths = ["tests"] diff --git a/examples/plugins/ntfy-notifier/tests/test_ntfy_notifier.py b/examples/plugins/ntfy-notifier/tests/test_ntfy_notifier.py new file mode 100644 index 00000000..04bfb814 --- /dev/null +++ b/examples/plugins/ntfy-notifier/tests/test_ntfy_notifier.py @@ -0,0 +1,21 @@ +import httpx +from ntfy_notifier import NtfyNotifier + +from ab_sdk.notify import NotifierSettings +from ab_sdk.testing import NotifierContract, create_plugin + + +def channel(status: int): + plugin, _ = create_plugin(NtfyNotifier, {"topic": "ab"}) + plugin.client_factory = lambda: httpx.AsyncClient( # type: ignore[method-assign] + transport=httpx.MockTransport(lambda request: httpx.Response(status)) + ) + return plugin.channel()(NotifierSettings()) + + +class TestChannel(NotifierContract): + def create(self): + return channel(200) + + def create_failing(self): + return channel(403) diff --git a/examples/plugins/template-rename/README.md b/examples/plugins/template-rename/README.md new file mode 100644 index 00000000..a666ce54 --- /dev/null +++ b/examples/plugins/template-rename/README.md @@ -0,0 +1,22 @@ +# 过滤器模板重命名（示例插件） + +自带过滤器的文件名模板，如 `{bangumi_name|sanitize} S{season|pad:2}E{episode|pad:2}`。演示： + +- `@provider(points.RENAME_STRATEGY)`：返回 `RenameStrategy` +- 渲染不出可用文件名时抛 `RenameSkipped`：宿主保留原文件名并通知，不计入熔断 +- 在 `field_validator` 里提前校验模板，写错的模板保存时就被拒绝 +- 契约套件 `RenameStrategyContract` + +## 安装 + +```bash +cp -r examples/plugins/template-rename config/plugins/local/ +``` + +在「设置 → 插件」中开启「允许未签名插件」并启用，再到「番剧管理设置 → 重命名方式」选择 `mini-template`。 + +## 测试 + +```bash +cd examples/plugins/template-rename && uv run pytest +``` diff --git a/examples/plugins/template-rename/plugin.toml b/examples/plugins/template-rename/plugin.toml new file mode 100644 index 00000000..ff18ad8f --- /dev/null +++ b/examples/plugins/template-rename/plugin.toml @@ -0,0 +1,9 @@ +[plugin] +id = "template-rename" +name = "过滤器模板重命名" +version = "0.1.0" +sdk = ">=0.5,<1" +entry = "template_rename:TemplateRename" +description = "示例插件：带自有过滤器的文件名模板（{field|filter:arg}）。在「番剧管理设置 → 重命名方式」选择 mini-template 启用。演示 rename_strategy" +authors = ["AutoBangumi"] +extension_points = ["rename_strategy"] diff --git a/examples/plugins/template-rename/pyproject.toml b/examples/plugins/template-rename/pyproject.toml new file mode 100644 index 00000000..fc87e0d2 --- /dev/null +++ b/examples/plugins/template-rename/pyproject.toml @@ -0,0 +1,13 @@ +# 只用于本地开发与测试，不会被打进插件包（ab-plugin pack 排除它） +[project] +name = "ab-plugin-template-rename" +version = "0.1.0" +requires-python = ">=3.13" +dependencies = ["autobangumi-sdk"] + +[dependency-groups] +dev = ["pytest>=8.0.0"] + +[tool.pytest.ini_options] +pythonpath = ["."] +testpaths = ["tests"] diff --git a/examples/plugins/template-rename/template_rename/__init__.py b/examples/plugins/template-rename/template_rename/__init__.py new file mode 100644 index 00000000..64d3ca98 --- /dev/null +++ b/examples/plugins/template-rename/template_rename/__init__.py @@ -0,0 +1,105 @@ +"""示例插件：带自有过滤器的重命名模板。 + +模板形如 ``{bangumi_name|sanitize} S{season|pad:2}E{episode|pad:2}``： +``{字段|过滤器|过滤器:参数}``，从左到右依次作用。渲染结果是文件名主体，字幕自动 +追加 ``.语言``，最后追加扩展名。与内置 ``template``（Jinja2）不同，这里自带 +过滤器表，不依赖第三方库。 + +模板写错在保存配置时就被拒绝（``field_validator``）。渲染出空名或含路径分隔符时 +抛出 :class:`RenameSkipped`：宿主保留原文件名并通知用户，绝不改成别的名字。 +""" + +import re +from collections.abc import Callable +from typing import Any + +from pydantic import BaseModel, Field, field_validator + +from ab_sdk import Plugin, points, provider +from ab_sdk.rename import RenameInput, RenameSkipped, RenameStrategy, pad + +# 文件名里各系统都不允许的字符 +_ILLEGAL = re.compile(r'[<>:"/\\|?*\x00-\x1f]') + +FILTERS: dict[str, Callable[..., Any]] = { + "pad": lambda v, width="2": pad(v, int(width)), + "sanitize": lambda v: " ".join(_ILLEGAL.sub(" ", str(v)).split()), + "short": lambda v, n="30": str(v)[: int(n)].rstrip(), + "upper": lambda v: str(v).upper(), + "lower": lambda v: str(v).lower(), +} +FIELDS = ( + "title", + "bangumi_name", + "season", + "episode", + "episode_type", + "group", + "kind", + "language", +) +_TOKEN = re.compile(r"\{([^{}]*)\}") + +SAMPLE = RenameInput( + kind="media", + media_path="[G] Sample - 01.mkv", + title="Sample", + bangumi_name="Sample (2024)", + season=1, + episode=1, + suffix=".mkv", + group="G", +) + + +def render(template: str, f: RenameInput) -> str: + def expand(match: re.Match[str]) -> str: + name, *chain = match.group(1).split("|") + if name not in FIELDS: + raise ValueError(f"未知字段 {name!r}，可用：{', '.join(FIELDS)}") + value: Any = getattr(f, name) or "" + for step in chain: + fn, _, arg = step.partition(":") + if fn not in FILTERS: + raise ValueError(f"未知过滤器 {fn!r}，可用：{', '.join(FILTERS)}") + value = FILTERS[fn](value, *([arg] if arg else [])) + return str(value) + + return _TOKEN.sub(expand, template) + + +class Options(BaseModel): + template: str = Field( + "{bangumi_name|sanitize} S{season|pad:2}E{episode|pad:2}", + title="文件名模板", + description=( + "{字段|过滤器:参数}。字段：" + "、".join(FIELDS) + "。过滤器：pad:位数、" + "sanitize（去掉非法字符）、short:长度、upper、lower" + ), + ) + + @field_validator("template") + @classmethod + def _check(cls, value: str) -> str: + render(value, SAMPLE) # 字段或过滤器写错时抛 ValueError，保存配置返回 422 + return value + + +class Strategy: + def __init__(self, template: str) -> None: + self.template = template + + def target_name(self, f: RenameInput) -> str: + stem = render(self.template, f).strip() + if not stem.strip(".") or "/" in stem or "\\" in stem: + raise RenameSkipped(f"模板渲染结果不可用：{stem!r}") + language = f".{f.language}" if f.kind == "subtitle" and f.language else "" + return f"{stem}{language}{f.suffix}" + + +class TemplateRename(Plugin[Options]): + config_model = Options + + @provider(points.RENAME_STRATEGY, id="mini-template") + def strategy(self) -> RenameStrategy: + return Strategy(self.config.template) diff --git a/examples/plugins/template-rename/tests/test_template_rename.py b/examples/plugins/template-rename/tests/test_template_rename.py new file mode 100644 index 00000000..60b360d5 --- /dev/null +++ b/examples/plugins/template-rename/tests/test_template_rename.py @@ -0,0 +1,58 @@ +import pytest +from pydantic import ValidationError +from template_rename import Options, Strategy, TemplateRename + +from ab_sdk.rename import RenameInput, RenameSkipped +from ab_sdk.testing import RenameStrategyContract, create_plugin + + +class TestStrategy(RenameStrategyContract): + def create(self): + plugin, _ = create_plugin(TemplateRename) + return plugin.strategy() + + +def media(**kwargs) -> RenameInput: + base = dict( + kind="media", + media_path="x.mkv", + title="Sample: Zero", + bangumi_name="Sample: Zero (2024)", + season=1, + episode=5, + suffix=".mkv", + ) + return RenameInput(**{**base, **kwargs}) + + +@pytest.mark.parametrize( + ("template", "f", "expected"), + [ + ( + "{bangumi_name|sanitize} S{season|pad:2}E{episode|pad:2}", + media(), + "Sample Zero (2024) S01E05.mkv", + ), + ("{title|short:6|upper}", media(), "SAMPLE.mkv"), + ( + "{title|sanitize}", + media(kind="subtitle", language="zh"), + "Sample Zero.zh.mkv", + ), + ("E{episode|pad:2}", media(episode=9.5), "E09.5.mkv"), + ], +) +def test_target_name_renders_filters(template, f, expected): + assert Strategy(template).target_name(f) == expected + + +@pytest.mark.parametrize("template", ["{nope}", "{title|bold}", "{episode|pad:x}"]) +def test_options_invalid_template_is_rejected(template): + with pytest.raises((ValidationError, ValueError)): + Options(template=template) + + +@pytest.mark.parametrize("template", ["{title}/{episode}", "{group}"]) +def test_target_name_unusable_result_skips_file(template): + with pytest.raises(RenameSkipped): + Strategy(template).target_name(media(title="a/b", group=None)) diff --git a/examples/plugins/webhook-on-event/README.md b/examples/plugins/webhook-on-event/README.md new file mode 100644 index 00000000..d6baab14 --- /dev/null +++ b/examples/plugins/webhook-on-event/README.md @@ -0,0 +1,21 @@ +# 事件 Webhook（示例插件） + +把选中的事件以 JSON POST 到一个 URL。演示： + +- `@subscribe("*")` 订阅全部事件，按配置过滤 +- `config_model`（WebUI 自动生成表单）与 `secret_field`（密码框、读取时掩码） +- 用 `ab_sdk.testing.create_plugin` 在不启动 AutoBangumi 的情况下测试 + +## 安装 + +```bash +cp -r examples/plugins/webhook-on-event config/plugins/local/ +``` + +在「设置 → 插件」中开启「允许未签名插件」，启用并填写地址。 + +## 测试 + +```bash +cd examples/plugins/webhook-on-event && uv run pytest +``` diff --git a/examples/plugins/webhook-on-event/plugin.toml b/examples/plugins/webhook-on-event/plugin.toml new file mode 100644 index 00000000..3519c584 --- /dev/null +++ b/examples/plugins/webhook-on-event/plugin.toml @@ -0,0 +1,9 @@ +[plugin] +id = "webhook-on-event" +name = "事件 Webhook" +version = "0.1.0" +sdk = ">=0.5,<1" +entry = "webhook_on_event:WebhookOnEvent" +description = "示例插件：把选中的事件以 JSON 形式 POST 到一个 URL，可选 HMAC 签名。演示 @subscribe、配置表单与秘密字段" +authors = ["AutoBangumi"] +permissions = ["network"] diff --git a/examples/plugins/webhook-on-event/pyproject.toml b/examples/plugins/webhook-on-event/pyproject.toml new file mode 100644 index 00000000..e379a3a7 --- /dev/null +++ b/examples/plugins/webhook-on-event/pyproject.toml @@ -0,0 +1,13 @@ +# 只用于本地开发与测试，不会被打进插件包（ab-plugin pack 排除它） +[project] +name = "ab-plugin-webhook-on-event" +version = "0.1.0" +requires-python = ">=3.13" +dependencies = ["autobangumi-sdk"] + +[dependency-groups] +dev = ["pytest>=8.0.0"] + +[tool.pytest.ini_options] +pythonpath = ["."] +testpaths = ["tests"] diff --git a/examples/plugins/webhook-on-event/tests/test_webhook_on_event.py b/examples/plugins/webhook-on-event/tests/test_webhook_on_event.py new file mode 100644 index 00000000..b52c9428 --- /dev/null +++ b/examples/plugins/webhook-on-event/tests/test_webhook_on_event.py @@ -0,0 +1,68 @@ +import asyncio +import hashlib +import hmac +import json + +import httpx +import pytest +from webhook_on_event import WebhookOnEvent + +from ab_sdk.events import OrganizedFile, TorrentOrganized +from ab_sdk.testing import create_plugin + +ORGANIZED = TorrentOrganized( + torrent_hash="abc", + bangumi_id=1, + files=(OrganizedFile(path="/a/b.mkv", kind="media"),), +) + + +def deliver(options: dict, status: int = 200): + """返回 (收到的请求, 发送函数)。""" + plugin, _ = create_plugin(WebhookOnEvent, options) + seen: list[httpx.Request] = [] + + def handler(request: httpx.Request) -> httpx.Response: + seen.append(request) + return httpx.Response(status) + + plugin.client_factory = lambda: httpx.AsyncClient( # type: ignore[method-assign] + transport=httpx.MockTransport(handler) + ) + return seen, lambda: asyncio.run(plugin.on_event(ORGANIZED)) + + +def test_on_event_selected_kind_posts_signed_json(): + seen, send = deliver({"url": "https://hook.test/x", "secret": "s3cret"}) + send() + [request] = seen + body = json.loads(request.content) + assert body["kind"] == "torrent.organized" + assert body["data"]["files"][0]["path"] == "/a/b.mkv" + expected = hmac.new(b"s3cret", request.content, hashlib.sha256).hexdigest() + assert request.headers["X-AB-Signature"] == expected + + +@pytest.mark.parametrize( + "options", + [ + {"url": ""}, + {"url": "https://hook.test/x", "events": ["rss_failure"]}, + ], +) +def test_on_event_unwanted_event_sends_nothing(options): + seen, send = deliver(options) + send() + assert seen == [] + + +def test_on_event_without_secret_omits_signature(): + seen, send = deliver({"url": "https://hook.test/x", "events": ["*"]}) + send() + assert "X-AB-Signature" not in seen[0].headers + + +def test_on_event_server_error_is_logged_not_raised(): + seen, send = deliver({"url": "https://hook.test/x"}, status=500) + send() + assert len(seen) == 1 diff --git a/examples/plugins/webhook-on-event/webhook_on_event/__init__.py b/examples/plugins/webhook-on-event/webhook_on_event/__init__.py new file mode 100644 index 00000000..22c46aac --- /dev/null +++ b/examples/plugins/webhook-on-event/webhook_on_event/__init__.py @@ -0,0 +1,68 @@ +"""示例插件：事件 Webhook。 + +订阅全部事件（``@subscribe("*")``），只把配置里列出的 ``kind`` POST 到 ``url``。 +请求体为 ``{"kind": ..., "data": {...}}``；配置了 ``secret`` 时，请求头 +``X-AB-Signature`` 带请求体的 HMAC-SHA256（十六进制）。 + +接收方不可达只记日志：webhook 是外部服务，它的故障不应让插件被熔断。 +""" + +import hashlib +import hmac +import json +from collections.abc import Callable +from dataclasses import asdict + +import httpx +from pydantic import BaseModel, Field + +from ab_sdk import Event, Plugin, secret_field, subscribe + + +class Options(BaseModel): + url: str = Field("", title="Webhook 地址", description="留空则什么也不发") + events: list[str] = Field( + default_factory=lambda: ["torrent.organized"], + title="事件", + description="要发送的事件 kind，如 torrent.organized、rss_failure；* 表示全部", + ) + secret: str = secret_field( + title="签名密钥", description="非空时在 X-AB-Signature 带请求体的 HMAC-SHA256" + ) + + +def encode(event: Event) -> bytes: + """事件是冻结 dataclass；``default=str`` 兜住 Path 等非 JSON 类型。""" + body = {"kind": event.kind, "data": asdict(event)} # type: ignore[call-overload] + return json.dumps(body, ensure_ascii=False, default=str).encode() + + +class WebhookOnEvent(Plugin[Options]): + config_model = Options + + # 测试可替换为带 MockTransport 的客户端 + client_factory: Callable[[], httpx.AsyncClient] = staticmethod( + lambda: httpx.AsyncClient(timeout=10) + ) + + def wants(self, event: Event) -> bool: + cfg = self.config + return bool(cfg.url) and ("*" in cfg.events or event.kind in cfg.events) + + @subscribe("*") + async def on_event(self, event: Event) -> None: + if not self.wants(event): + return + body = encode(event) + headers = {"Content-Type": "application/json"} + if self.config.secret: + mac = hmac.new(self.config.secret.encode(), body, hashlib.sha256) + headers["X-AB-Signature"] = mac.hexdigest() + try: + async with self.client_factory() as client: + resp = await client.post(self.config.url, content=body, headers=headers) + except httpx.HTTPError as e: + self.ctx.log.warning("Webhook 请求失败：%s", e) + return + if not resp.is_success: + self.ctx.log.warning("Webhook 返回 HTTP %s", resp.status_code) diff --git a/scripts/test_example_plugins.sh b/scripts/test_example_plugins.sh new file mode 100755 index 00000000..0cb3873e --- /dev/null +++ b/scripts/test_example_plugins.sh @@ -0,0 +1,14 @@ +#!/usr/bin/env bash +# 逐个示例插件：校验清单，再跑它自己的测试（与插件作者在示例目录里 `uv run pytest` 一致）。 +# 在仓库根目录运行；CI 与本地共用。 +set -euo pipefail +root="$(cd "$(dirname "$0")/.." && pwd)" +cd "$root/backend" +for dir in "$root"/examples/plugins/*/; do + name="$(basename "$dir")" + echo "== $name" + uv run ab-plugin validate "$dir" + if [ -d "$dir/tests" ]; then + uv run pytest -q -c "$dir/pyproject.toml" --rootdir "$dir" -p no:cacheprovider "$dir/tests" + fi +done