feat(4.0-P4): 内置插件 media-server-refresh 在整理完成后刷新 Jellyfin / Emby / Plex 媒体库

订阅 torrent.organized（含重命名方式为 none 的种子），等待 delay 秒把期间的
事件合并成一次刷新请求。未填写服务器地址或 API Key 时不做任何事，因此与
另一份 P4 实现相同，内置默认启用。

backend/src/module/plugins/builtin/media-server-refresh/media_server_refresh.py

@@ -0,0 +1,107 @@
+"""内置插件：种子整理完成后通知媒体服务器刷新媒体库（订阅 ``torrent.organized``）。
+
+收到第一个事件后等待 ``delay`` 秒，期间整理完成的所有种子合并成一次刷新请求。
+未填写服务器地址或 API Key 时什么也不做，所以默认启用也无副作用。
+
+``torrent.organized`` 投递为「至少一次」：重命名方式为 none 的种子每次进程
+重启后会再发布一次，已配置时重启后最多多出一次（合并后的）刷新。
+"""
+
+import asyncio
+from collections.abc import Callable
+from typing import Literal
+
+import httpx
+from pydantic import BaseModel, Field
+
+from ab_sdk import Plugin, PluginContext, secret_field, subscribe
+from ab_sdk.events import TorrentOrganized
+
+ServerType = Literal["jellyfin", "emby", "plex"]
+
+
+class Options(BaseModel):
+    server: ServerType = Field("jellyfin", title="服务器类型")
+    url: str = Field(
+        "",
+        title="服务器地址",
+        description="如 http://192.168.1.10:8096（Plex 默认端口 32400）；留空则不刷新",
+    )
+    api_key: str = secret_field(
+        title="API Key",
+        description="Jellyfin / Emby 的 API Key，或 Plex 的 X-Plex-Token",
+    )
+    delay: float = Field(
+        30,
+        ge=0,
+        le=3600,
+        title="延迟（秒）",
+        description="收到整理完成事件后等待多久再刷新，期间的事件合并为一次刷新",
+    )
+
+
+def build_request(options: Options) -> tuple[str, str, dict[str, str]]:
+    """返回 (HTTP 方法, URL, 请求头)。"""
+    base = options.url.strip().rstrip("/")
+    key = options.api_key
+    if options.server == "plex":
+        return "GET", f"{base}/library/sections/all/refresh", {"X-Plex-Token": key}
+    if options.server == "emby":
+        return "POST", f"{base}/emby/Library/Refresh", {"X-Emby-Token": key}
+    return (
+        "POST",
+        f"{base}/Library/Refresh",
+        {"Authorization": f'MediaBrowser Token="{key}"'},
+    )
+
+
+class MediaServerRefresh(Plugin[Options]):
+    config_model = Options
+
+    def __init__(self, ctx: PluginContext) -> None:
+        super().__init__(ctx)
+        self._pending: asyncio.Task[None] | None = None
+        # 测试可替换为带 MockTransport 的客户端
+        self.client_factory: Callable[[], httpx.AsyncClient] = lambda: (
+            httpx.AsyncClient(timeout=10)
+        )
+
+    @property
+    def configured(self) -> bool:
+        return bool(self.config.url.strip() and self.config.api_key)
+
+    async def teardown(self) -> None:
+        if self._pending is not None and not self._pending.done():
+            self._pending.cancel()
+            try:
+                await self._pending
+            except asyncio.CancelledError:
+                pass
+        self._pending = None
+
+    @subscribe(TorrentOrganized.kind)
+    async def on_organized(self, event: TorrentOrganized) -> None:
+        if not self.configured:
+            return
+        if self._pending is None or self._pending.done():
+            self.ctx.log.debug("种子 %s 整理完成，计划刷新媒体库", event.torrent_hash)
+            self._pending = asyncio.create_task(self._refresh_later())
+
+    async def _refresh_later(self) -> None:
+        await asyncio.sleep(self.config.delay)
+        try:
+            await self.refresh()
+        except Exception as e:
+            self.ctx.log.warning("刷新媒体库失败：%s", e)
+
+    async def refresh(self) -> bool:
+        method, url, headers = build_request(self.config)
+        async with self.client_factory() as client:
+            resp = await client.request(method, url, headers=headers)
+        if resp.is_success:
+            self.ctx.log.info("已请求 %s 刷新媒体库", self.config.server)
+            return True
+        self.ctx.log.warning(
+            "%s 刷新媒体库失败：HTTP %s", self.config.server, resp.status_code
+        )
+        return False

backend/src/module/plugins/builtin/media-server-refresh/plugin.toml

@@ -0,0 +1,9 @@
+[plugin]
+id = "media-server-refresh"
+name = "媒体库刷新"
+version = "1.0.0"
+sdk = ">=0.4,<1"
+entry = "media_server_refresh:MediaServerRefresh"
+description = "种子整理完成后通知 Jellyfin / Emby / Plex 刷新媒体库；填写服务器地址与 API Key 后生效"
+authors = ["AutoBangumi"]
+permissions = ["network"]

backend/src/test/test_media_server_refresh.py

@@ -0,0 +1,136 @@
+"""内置插件 media-server-refresh：订阅 torrent.organized，合并后请求媒体服务器刷新。"""
+
+import asyncio
+import logging
+
+import httpx
+import pytest
+
+from ab_sdk.events import OrganizedFile, TorrentOrganized
+from ab_sdk.testing import create_plugin
+from module.plugin.loader import BUILTIN_ROOT, discover
+
+
+def load_candidate():
+    candidates, errors = discover(
+        local_root=BUILTIN_ROOT / "__missing__", entry_point_group="ab-test-none"
+    )
+    assert errors == []
+    return {c.manifest.id: c for c in candidates}["media-server-refresh"]
+
+
+def organized(n: int = 1) -> TorrentOrganized:
+    return TorrentOrganized(
+        torrent_hash=f"h{n}",
+        bangumi_id=1,
+        files=(OrganizedFile(f"/dl/A S01E0{n}.mkv", "media"),),
+    )
+
+
+def make_plugin(options, handler=None):
+    plugin, _ = create_plugin(
+        load_candidate().load(), options, plugin_id="media-server-refresh"
+    )
+    requests: list[httpx.Request] = []
+
+    def record(request: httpx.Request) -> httpx.Response:
+        requests.append(request)
+        return handler(request) if handler else httpx.Response(204)
+
+    plugin.client_factory = lambda: httpx.AsyncClient(
+        transport=httpx.MockTransport(record)
+    )
+    return plugin, requests
+
+
+def test_media_server_refresh_builtin_enabled_by_default_with_secret_key():
+    candidate = load_candidate()
+    assert candidate.source == "builtin"
+    assert candidate.manifest.default_enabled is True
+    schema = candidate.load().config_model.model_json_schema()
+    assert schema["properties"]["api_key"]["secret"] is True
+
+
+@pytest.mark.parametrize(
+    "options", [{}, {"url": "http://jf:8096"}, {"api_key": "k"}, {"url": " "}]
+)
+async def test_on_organized_unconfigured_is_noop(options):
+    plugin, requests = make_plugin(options)
+    await plugin.on_organized(organized())
+    assert plugin._pending is None
+    await plugin.teardown()
+    assert requests == []
+
+
+@pytest.mark.parametrize(
+    ("server", "method", "url", "header", "value"),
+    [
+        (
+            "jellyfin",
+            "POST",
+            "http://srv:8096/Library/Refresh",
+            "authorization",
+            'MediaBrowser Token="k3y"',
+        ),
+        ("emby", "POST", "http://srv:8096/emby/Library/Refresh", "x-emby-token", "k3y"),
+        (
+            "plex",
+            "GET",
+            "http://srv:8096/library/sections/all/refresh",
+            "x-plex-token",
+            "k3y",
+        ),
+    ],
+)
+async def test_on_organized_burst_coalesces_into_one_refresh(
+    server, method, url, header, value
+):
+    plugin, requests = make_plugin(
+        {"server": server, "url": "http://srv:8096/", "api_key": "k3y", "delay": 0.05}
+    )
+    for n in range(1, 4):
+        await plugin.on_organized(organized(n))
+    assert plugin._pending is not None
+    await asyncio.wait_for(plugin._pending, 1)
+
+    assert len(requests) == 1
+    assert requests[0].method == method
+    assert str(requests[0].url) == url
+    assert requests[0].headers[header] == value
+    # 刷新完成后的新事件会再排一次
+    await plugin.on_organized(organized(4))
+    await asyncio.wait_for(plugin._pending, 1)
+    assert len(requests) == 2
+
+
+async def test_refresh_http_error_returns_false():
+    plugin, requests = make_plugin(
+        {"url": "http://srv", "api_key": "k"}, handler=lambda r: httpx.Response(401)
+    )
+    assert await plugin.refresh() is False
+    assert len(requests) == 1
+
+
+async def test_on_organized_connection_error_logged_without_secret(caplog):
+    def explode(request):
+        raise httpx.ConnectError("refused")
+
+    plugin, _ = make_plugin(
+        {"url": "http://srv", "api_key": "s3cr3t-key", "delay": 0}, handler=explode
+    )
+    with caplog.at_level(logging.WARNING):
+        await plugin.on_organized(organized())
+        assert plugin._pending is not None
+        await asyncio.wait_for(plugin._pending, 1)
+    assert "刷新媒体库失败" in caplog.text
+    assert "s3cr3t-key" not in caplog.text
+
+
+async def test_teardown_pending_refresh_cancelled():
+    plugin, requests = make_plugin({"url": "http://srv", "api_key": "k", "delay": 60})
+    await plugin.on_organized(organized())
+    pending = plugin._pending
+    assert pending is not None
+    await plugin.teardown()
+    assert pending.cancelled()
+    assert requests == []

backend/vulture_whitelist.py

@@ -85,5 +85,6 @@ _.pn_strategy
 _.advance_strategy
 _.template_strategy
 HardlinkPlugin
-_.on_organized
+MediaServerRefresh
+_.on_organized  # hardlink 与 media-server-refresh 共用
 _.api  # 插件 HTTP 路由，宿主挂载到 /api/v1/plugins/<id>/