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