From d51e77c37fecedb713f8887d7b0cc6a7fbbf7bcf Mon Sep 17 00:00:00 2001 From: Estrella Pan Date: Thu, 8 Oct 2026 11:32:38 +0200 Subject: [PATCH] =?UTF-8?q?feat(4.0-P4):=20=E5=86=85=E7=BD=AE=E6=8F=92?= =?UTF-8?q?=E4=BB=B6=20media-server-refresh=20=E5=9C=A8=E6=95=B4=E7=90=86?= =?UTF-8?q?=E5=AE=8C=E6=88=90=E5=90=8E=E5=88=B7=E6=96=B0=20Jellyfin=20/=20?= =?UTF-8?q?Emby=20/=20Plex=20=E5=AA=92=E4=BD=93=E5=BA=93?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 订阅 torrent.organized（含重命名方式为 none 的种子），等待 delay 秒把期间的 事件合并成一次刷新请求。未填写服务器地址或 API Key 时不做任何事，因此与 另一份 P4 实现相同，内置默认启用。 --- .../media_server_refresh.py | 107 ++++++++++++++ .../builtin/media-server-refresh/plugin.toml | 9 ++ backend/src/test/test_media_server_refresh.py | 136 ++++++++++++++++++ backend/vulture_whitelist.py | 3 +- 4 files changed, 254 insertions(+), 1 deletion(-) create mode 100644 backend/src/module/plugins/builtin/media-server-refresh/media_server_refresh.py create mode 100644 backend/src/module/plugins/builtin/media-server-refresh/plugin.toml create mode 100644 backend/src/test/test_media_server_refresh.py diff --git a/backend/src/module/plugins/builtin/media-server-refresh/media_server_refresh.py b/backend/src/module/plugins/builtin/media-server-refresh/media_server_refresh.py new file mode 100644 index 00000000..bb5b5cb5 --- /dev/null +++ b/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 diff --git a/backend/src/module/plugins/builtin/media-server-refresh/plugin.toml b/backend/src/module/plugins/builtin/media-server-refresh/plugin.toml new file mode 100644 index 00000000..ea6d13ad --- /dev/null +++ b/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"] diff --git a/backend/src/test/test_media_server_refresh.py b/backend/src/test/test_media_server_refresh.py new file mode 100644 index 00000000..e58af2ba --- /dev/null +++ b/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 == [] diff --git a/backend/vulture_whitelist.py b/backend/vulture_whitelist.py index 261b74c2..f0ee0551 100644 --- a/backend/vulture_whitelist.py +++ b/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//