From 34526246b791d706b46c25304dcd5afebbb92f0b Mon Sep 17 00:00:00 2001 From: Estrella Pan Date: Thu, 8 Oct 2026 16:46:13 +0200 Subject: [PATCH] =?UTF-8?q?fix(4.0):=20=E5=AA=92=E4=BD=93=E5=BA=93?= =?UTF-8?q?=E5=88=B7=E6=96=B0=E5=9C=A8=E7=A1=AC=E9=93=BE=E6=8E=A5=E6=94=BE?= =?UTF-8?q?=E5=A5=BD=E6=96=87=E4=BB=B6=E5=90=8E=E5=86=8D=E5=88=B7=E6=96=B0?= =?UTF-8?q?=E4=B8=80=E6=AC=A1=EF=BC=8C=E5=88=B7=E6=96=B0=E6=9C=9F=E9=97=B4?= =?UTF-8?q?=E7=9A=84=E4=BA=8B=E4=BB=B6=E4=B8=8D=E5=86=8D=E4=B8=A2=E5=A4=B1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit media-server-refresh 只按第一个 torrent.organized 起算 delay，不等待 同一事件的 hardlink 订阅者：跨盘复制慢于 delay 时刷新早于文件到位。 hardlink 新放入文件后发布 hardlink.linked（普通事件，不是通知）， media-server-refresh 同时订阅它。刷新请求发出后到达的事件原先被忽略， 现在这次刷新之后再排一次。 --- .../builtin/hardlink/hardlink_plugin.py | 16 +++++++- .../media_server_refresh.py | 33 +++++++++++---- backend/src/test/test_hardlink.py | 40 ++++++++++++++----- backend/src/test/test_media_server_refresh.py | 40 +++++++++++++++++++ backend/vulture_whitelist.py | 1 + docs/config/manager.md | 2 +- docs/dev/plugins/builtin.md | 2 +- docs/en/config/manager.md | 2 +- docs/en/dev/plugins/builtin.md | 2 +- docs/ja/config/manager.md | 2 +- docs/ja/dev/plugins/builtin.md | 2 +- ...26-10-08-plugin-architecture-4.0-design.md | 2 +- 12 files changed, 118 insertions(+), 26 deletions(-) diff --git a/backend/src/module/plugins/builtin/hardlink/hardlink_plugin.py b/backend/src/module/plugins/builtin/hardlink/hardlink_plugin.py index 5e0bea6e..10aa449f 100644 --- a/backend/src/module/plugins/builtin/hardlink/hardlink_plugin.py +++ b/backend/src/module/plugins/builtin/hardlink/hardlink_plugin.py @@ -26,7 +26,7 @@ from fastapi import APIRouter from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator from ab_sdk import Plugin, points, provider, subscribe -from ab_sdk.events import SystemEvent, TorrentOrganized +from ab_sdk.events import Event, SystemEvent, TorrentOrganized Status = Literal["linked", "exists", "conflict", "failed"] @@ -102,6 +102,16 @@ class HardlinkFailed(SystemEvent): return ("硬链接未完成", f"种子：{self.torrent_hash}\n{self.files}") +@dataclass(frozen=True, slots=True) +class HardlinkLinked(Event): + """种子有文件新放入了媒体库。跨盘复制可能比媒体库刷新的延迟更久， + media-server-refresh 订阅它，在文件放好后再刷新一次。""" + + kind: ClassVar[str] = "hardlink.linked" + + torrent_hash: str + + def _identity(path: Path) -> list[int]: """文件身份（不跟随软链接）。含大小与修改时间：旧文件删除后 inode 可能被复用。""" st = os.lstat(path) @@ -216,12 +226,16 @@ class HardlinkPlugin(Plugin[Options]): @subscribe(TorrentOrganized.kind, timeout=LINK_TIMEOUT) async def on_organized(self, event: TorrentOrganized) -> None: problems: list[str] = [] + linked = False for f in event.files: status, reason = await self.link( Path(self.to_local(f.path, event.downloader_id)) ) + linked = linked or status == "linked" if status in ("conflict", "failed"): problems.append(f"{f.path}：{reason}") + if linked: + self.ctx.bus.publish(HardlinkLinked(torrent_hash=event.torrent_hash)) if problems: self.ctx.log.warning("种子 %s 未完成链接：%s", event.torrent_hash, problems) self.ctx.bus.publish( 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 index c8bbf8b7..029c6d14 100644 --- 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 @@ -1,6 +1,7 @@ """内置插件：种子整理完成后通知媒体服务器刷新媒体库（订阅 ``torrent.organized``）。 收到第一个事件后等待 ``delay`` 秒，期间整理完成的所有种子合并成一次刷新请求。 +同时订阅 hardlink 插件的 ``hardlink.linked``：文件放入媒体库后再刷新一次。 未填写服务器地址或 API Key 时什么也不做，所以默认启用也无副作用。 ``torrent.organized`` 投递为「至少一次」：下载器中已整理的种子每次进程重启后 @@ -15,7 +16,7 @@ import httpx from pydantic import BaseModel, Field from ab_sdk import Plugin, PluginContext, secret_field, subscribe -from ab_sdk.events import TorrentOrganized +from ab_sdk.events import Event, TorrentOrganized ServerType = Literal["jellyfin", "emby", "plex"] @@ -61,6 +62,8 @@ class MediaServerRefresh(Plugin[Options]): def __init__(self, ctx: PluginContext) -> None: super().__init__(ctx) self._pending: asyncio.Task[None] | None = None + # 刷新请求发出后又有新事件：这次刷新之后再排一次 + self._again = False # 测试可替换为带 MockTransport 的客户端 self.client_factory: Callable[[], httpx.AsyncClient] = lambda: ( httpx.AsyncClient(timeout=10) @@ -81,18 +84,34 @@ class MediaServerRefresh(Plugin[Options]): @subscribe(TorrentOrganized.kind) async def on_organized(self, event: TorrentOrganized) -> None: + self._schedule(event.kind) + + # hardlink 插件把文件放入媒体库后发布；跨盘复制可能比 delay 更久， + # 放好后再刷新一次（期间的事件同样合并） + @subscribe("hardlink.linked") + async def on_linked(self, event: Event) -> None: + self._schedule(event.kind) + + def _schedule(self, kind: str) -> None: if not self.configured: return if self._pending is None or self._pending.done(): - self.ctx.log.debug("种子 %s 整理完成，计划刷新媒体库", event.torrent_hash) + self.ctx.log.debug("收到 %s，计划刷新媒体库", kind) self._pending = asyncio.create_task(self._refresh_later()) + else: + self._again = True 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) + again = True + while again: + await asyncio.sleep(self.config.delay) + # 等待期间到达的事件由这次刷新覆盖 + self._again = False + try: + await self.refresh() + except Exception as e: + self.ctx.log.warning("刷新媒体库失败：%s", e) + again = self._again async def refresh(self) -> bool: method, url, headers = build_request(self.config) diff --git a/backend/src/test/test_hardlink.py b/backend/src/test/test_hardlink.py index 69175a09..74b75ce3 100644 --- a/backend/src/test/test_hardlink.py +++ b/backend/src/test/test_hardlink.py @@ -60,6 +60,10 @@ def organized(*names: str, downloader: str = "default") -> TorrentOrganized: ) +def failures(ctx) -> list: + return [e for e in ctx.bus.published if e.kind == "hardlink.failed"] + + def exdev(src, dst): raise OSError(errno.EXDEV, "Invalid cross-device link") @@ -90,7 +94,7 @@ async def test_on_organized_maps_longest_prefix_of_same_downloader_and_links( assert os.path.samefile(src, library / SEASON / src.name) assert os.path.samefile(sub, library / SEASON / sub.name) - assert ctx.bus.published == [] + assert failures(ctx) == [] @pytest.mark.parametrize("mode", ["copy", "symlink", "skip"]) @@ -105,12 +109,12 @@ async def test_on_organized_cross_device_applies_policy(roots, monkeypatch, mode dst = library / SEASON / src.name if mode == "skip": assert not dst.exists() - [event] = ctx.bus.published + [event] = failures(ctx) assert event.kind == "hardlink.failed" and src.name in event.files return assert dst.read_text() == "v1" assert dst.is_symlink() is (mode == "symlink") - assert ctx.bus.published == [] + assert failures(ctx) == [] async def test_on_organized_redelivery_does_not_copy_again(roots, monkeypatch): @@ -130,7 +134,7 @@ async def test_on_organized_redelivery_does_not_copy_again(roots, monkeypatch): await plugin.on_organized(organized(src.name)) assert len(copies) == 1 - assert ctx.bus.published == [] + assert failures(ctx) == [] async def test_on_organized_interrupted_copy_leaves_no_target(roots, monkeypatch): @@ -152,7 +156,7 @@ async def test_on_organized_interrupted_copy_leaves_no_target(roots, monkeypatch assert (library / SEASON / src.name).read_text() == "v1" assert [p.name for p in (library / SEASON).iterdir()] == [src.name] - [failed] = ctx.bus.published + [failed] = failures(ctx) assert "No space left" in failed.files @@ -170,7 +174,7 @@ async def test_on_organized_revision_upgrade_replaces_own_link(roots): dst = library / SEASON / src.name assert os.path.samefile(src, dst) and dst.read_text() == "v2" assert [p.name for p in dst.parent.iterdir()] == [src.name] - assert ctx.bus.published == [] + assert failures(ctx) == [] async def test_on_organized_problem_files_kept_and_notified_once(roots, tmp_path): @@ -189,7 +193,7 @@ async def test_on_organized_problem_files_kept_and_notified_once(roots, tmp_path await plugin.on_organized(organized(taken.name, "outside/x.mkv")) assert (library / SEASON / taken.name).read_text() == "user file" - [event] = ctx.bus.published + [event] = failures(ctx) assert event.kind == "hardlink.failed" and event.torrent_hash == "h1" assert taken.name in event.files and "x.mkv" in event.files @@ -213,7 +217,7 @@ def test_backfill_route_links_existing_files_and_counts(roots): downloads / SEASON / "new.zh.ass", library / SEASON / "new.zh.ass" ) assert not (library / SEASON / "note.nfo").exists() - assert ctx.bus.published == [] + assert failures(ctx) == [] @pytest.mark.parametrize( @@ -250,7 +254,7 @@ async def test_on_organized_unrecorded_own_copy_adopted_and_upgraded( await plugin.on_organized(organized(src.name)) assert dst.read_text() == "v2" - assert ctx.bus.published == [] + assert failures(ctx) == [] async def test_on_organized_user_file_at_former_link_kept(roots): @@ -268,7 +272,7 @@ async def test_on_organized_user_file_at_former_link_kept(roots): await plugin.on_organized(organized(src.name)) assert dst.read_text() == "user file" - [event] = ctx.bus.published + [event] = failures(ctx) assert event.kind == "hardlink.failed" and src.name in event.files @@ -317,4 +321,18 @@ async def test_link_target_deleted_by_user_restored_only_by_backfill( await plugin.on_organized(organized(src.name)) assert dst.exists() is via_backfill - assert ctx.bus.published == [] + assert failures(ctx) == [] + + +async def test_on_organized_publishes_linked_only_when_file_placed(roots): + downloads, library = roots + plugin, ctx = make_plugin(roots) + src = write(downloads / SEASON / "Anime S01E01.mkv", "v1") + + await plugin.on_organized(organized(src.name)) + await plugin.on_organized(organized(src.name)) + + # 媒体库刷新据此在文件放好后再刷新一次；已链接的重复投递不再发布 + assert [(e.kind, e.torrent_hash) for e in ctx.bus.published] == [ + ("hardlink.linked", "h1") + ] diff --git a/backend/src/test/test_media_server_refresh.py b/backend/src/test/test_media_server_refresh.py index e58af2ba..8f800e8a 100644 --- a/backend/src/test/test_media_server_refresh.py +++ b/backend/src/test/test_media_server_refresh.py @@ -2,11 +2,13 @@ import asyncio import logging +from types import SimpleNamespace import httpx import pytest from ab_sdk.events import OrganizedFile, TorrentOrganized +from ab_sdk.hooks import SUBSCRIBE_ATTR from ab_sdk.testing import create_plugin from module.plugin.loader import BUILTIN_ROOT, discover @@ -134,3 +136,41 @@ async def test_teardown_pending_refresh_cancelled(): await plugin.teardown() assert pending.cancelled() assert requests == [] + + +async def test_on_organized_during_refresh_request_refreshes_again(): + plugin, _ = make_plugin({"url": "http://srv", "api_key": "k", "delay": 0}) + requests: list[httpx.Request] = [] + + async def handler(request: httpx.Request) -> httpx.Response: + requests.append(request) + if len(requests) == 1: + # 刷新请求已发出、尚未返回时又有种子整理完成 + await plugin.on_organized(organized(2)) + return httpx.Response(204) + + plugin.client_factory = lambda: httpx.AsyncClient( + transport=httpx.MockTransport(handler) + ) + await plugin.on_organized(organized(1)) + assert plugin._pending is not None + await asyncio.wait_for(plugin._pending, 1) + + assert len(requests) == 2 + + +async def test_hardlink_linked_event_schedules_refresh(): + plugin, requests = make_plugin({"url": "http://srv", "api_key": "k", "delay": 0}) + # 硬链接（跨盘复制可能比 delay 更久）放好文件后再刷新一次 + [handler] = [ + getattr(plugin, name) + for name, member in vars(type(plugin)).items() + if getattr(getattr(member, SUBSCRIBE_ATTR, None), "kind", None) + == "hardlink.linked" + ] + + await handler(SimpleNamespace(kind="hardlink.linked", torrent_hash="h1")) + assert plugin._pending is not None + await asyncio.wait_for(plugin._pending, 1) + + assert len(requests) == 1 diff --git a/backend/vulture_whitelist.py b/backend/vulture_whitelist.py index f11e0863..3197a6f7 100644 --- a/backend/vulture_whitelist.py +++ b/backend/vulture_whitelist.py @@ -89,4 +89,5 @@ _.template_strategy HardlinkPlugin MediaServerRefresh _.on_organized # hardlink 与 media-server-refresh 共用 +_.on_linked # media-server-refresh 订阅 hardlink.linked _.api # 插件 HTTP 路由，宿主挂载到 /api/v1/plugins// diff --git a/docs/config/manager.md b/docs/config/manager.md index 5f4c30e3..ea7d13e7 100644 --- a/docs/config/manager.md +++ b/docs/config/manager.md @@ -91,7 +91,7 @@ ## 媒体库刷新 -内置插件「媒体库刷新」（`media-server-refresh`）在种子整理完成后请求 Jellyfin、Emby 或 Plex 刷新媒体库。在 **设置 → 插件 → 媒体库刷新** 中选择服务器类型，填写服务器地址与 API Key（Plex 为 `X-Plex-Token`）后生效；未填写时不做任何事。收到事件后等待「延迟」秒（默认 30），期间整理完成的种子合并为一次刷新。 +内置插件「媒体库刷新」（`media-server-refresh`）在种子整理完成后请求 Jellyfin、Emby 或 Plex 刷新媒体库。在 **设置 → 插件 → 媒体库刷新** 中选择服务器类型，填写服务器地址与 API Key（Plex 为 `X-Plex-Token`）后生效；未填写时不做任何事。收到事件后等待「延迟」秒（默认 30），期间整理完成的种子合并为一次刷新；刷新请求发出后到达的事件会再排一次刷新。启用「硬链接到媒体库」时，文件放入媒体库后会再刷新一次，跨盘复制慢于延迟时新剧集也能出现。 [1]: https://www.autobangumi.org/faq/#download-path [2]: https://www.autobangumi.org/faq/#file-renaming diff --git a/docs/dev/plugins/builtin.md b/docs/dev/plugins/builtin.md index 0d1da129..21bfed2b 100644 --- a/docs/dev/plugins/builtin.md +++ b/docs/dev/plugins/builtin.md @@ -23,7 +23,7 @@ ## media-server-refresh -订阅 `torrent.organized`。收到第一个事件后等待 `delay` 秒，期间整理完成的种子合并成一次刷新请求。 +订阅 `torrent.organized` 与 `hardlink` 发布的 `hardlink.linked`（文件新放入媒体库）。收到第一个事件后等待 `delay` 秒，期间的事件合并成一次刷新请求；刷新请求发出后到达的事件再排一次。 | 选项 | 说明 | | --- | --- | diff --git a/docs/en/config/manager.md b/docs/en/config/manager.md index 77282888..119f6ede 100644 --- a/docs/en/config/manager.md +++ b/docs/en/config/manager.md @@ -75,7 +75,7 @@ Hard links cannot cross file systems. In Docker, keep the download folder and th ## Media server refresh -The built-in `media-server-refresh` plugin asks Jellyfin, Emby or Plex to refresh the library after a torrent is organized. Under **Settings → Plugins → media-server-refresh**, select the server type and enter the server URL and API key (for Plex, the `X-Plex-Token`). Without them the plugin does nothing. After an event it waits `delay` seconds (default 30) and sends one refresh for all torrents organized in that time. +The built-in `media-server-refresh` plugin asks Jellyfin, Emby or Plex to refresh the library after a torrent is organized. Under **Settings → Plugins → media-server-refresh**, select the server type and enter the server URL and API key (for Plex, the `X-Plex-Token`). Without them the plugin does nothing. After an event it waits `delay` seconds (default 30) and sends one refresh for all torrents organized in that time. Events that arrive after the refresh request is sent cause one more refresh. When the hardlink plugin is enabled, the plugin refreshes again after the files are in the library, so a copy across disks that takes longer than the delay does not hide the new episode. ## `config.json` diff --git a/docs/en/dev/plugins/builtin.md b/docs/en/dev/plugins/builtin.md index c22bfe08..5f722d84 100644 --- a/docs/en/dev/plugins/builtin.md +++ b/docs/en/dev/plugins/builtin.md @@ -23,7 +23,7 @@ Adds an include filter. The rule already has an exclude filter. Set a list of re ## media-server-refresh -Subscribes to `torrent.organized`. After the first event, it waits `delay` seconds. The torrents that finish in this time share one refresh request. +Subscribes to `torrent.organized` and to `hardlink.linked`, which `hardlink` publishes when it places new files in the library. After the first event, it waits `delay` seconds. The events in this time share one refresh request. An event that arrives after the request is sent causes one more refresh. | Option | Description | | --- | --- | diff --git a/docs/ja/config/manager.md b/docs/ja/config/manager.md index 836ee088..346258df 100644 --- a/docs/ja/config/manager.md +++ b/docs/ja/config/manager.md @@ -73,7 +73,7 @@ ## メディアサーバーの更新 -内蔵プラグイン `media-server-refresh` は、Torrent の整理後に Jellyfin・Emby・Plex へライブラリの更新を要求します。**設定 → プラグイン → media-server-refresh** でサーバー種別を選び、サーバー URL と API Key（Plex は `X-Plex-Token`）を入力すると有効になります。未入力の間は何もしません。イベントを受け取ると `delay` 秒（既定 30）待ち、その間に整理された Torrent をまとめて 1 回だけ更新します。 +内蔵プラグイン `media-server-refresh` は、Torrent の整理後に Jellyfin・Emby・Plex へライブラリの更新を要求します。**設定 → プラグイン → media-server-refresh** でサーバー種別を選び、サーバー URL と API Key（Plex は `X-Plex-Token`）を入力すると有効になります。未入力の間は何もしません。イベントを受け取ると `delay` 秒（既定 30）待ち、その間に整理された Torrent をまとめて 1 回だけ更新します。更新リクエストの送信後に届いたイベントは、もう 1 回の更新になります。ハードリンクプラグインが有効な場合は、ファイルがライブラリに入った後にもう一度更新します。ディスクをまたぐコピーが遅延より長くかかっても、新しいエピソードが表示されます。 ## `config.json` diff --git a/docs/ja/dev/plugins/builtin.md b/docs/ja/dev/plugins/builtin.md index 338eaa88..d9ac46c6 100644 --- a/docs/ja/dev/plugins/builtin.md +++ b/docs/ja/dev/plugins/builtin.md @@ -23,7 +23,7 @@ ## media-server-refresh -`torrent.organized` を購読します。最初のイベントを受け取ってから `delay` 秒待ち、その間に整理が完了したトレントを 1 回の更新リクエストにまとめます。 +`torrent.organized` と、`hardlink` が新しいファイルをライブラリに置いたときに発行する `hardlink.linked` を購読します。最初のイベントを受け取ってから `delay` 秒待ち、その間のイベントを 1 回の更新リクエストにまとめます。リクエスト送信後に届いたイベントは、もう 1 回の更新になります。 | オプション | 説明 | | --- | --- | diff --git a/docs/plans/2026-10-08-plugin-architecture-4.0-design.md b/docs/plans/2026-10-08-plugin-architecture-4.0-design.md index 8d884781..b19b46fc 100644 --- a/docs/plans/2026-10-08-plugin-architecture-4.0-design.md +++ b/docs/plans/2026-10-08-plugin-architecture-4.0-design.md @@ -91,7 +91,7 @@ - P4 有两份独立实现。另一个会话的版本（提交 a0706513、12e7e2f5，基于旧 P5 0b6de315）曾推到 `refactor/4.0-p4-organize`，现保留为 `refactor/4.0-p4-organize-cloud`（head 544628f6）。本分支保留本节上文的设计，因为它符合已确认的决策：`pn` / `advance` / `template` 都在内置插件 `rename` 中，坏模板跳过文件并通知、不退回 `pn`，另有 `hardlink`。另一份的 `rename-template` 插件、`BangumiLink`、`ab_sdk.organize` 与由 `PluginManager` 设置进程级总线的做法没有移植。 - 从另一份移植并按本分支改写的内容： - - **内置插件 `media-server-refresh`**：订阅 `torrent.organized`（另一份订阅 `file.renamed`；前者在重命名方式为 `none` 时也会发布），等待 `delay` 秒，把期间的事件合并成一次 Jellyfin / Emby / Plex 刷新请求。未填写地址或 API Key 时不做任何事，所以与另一份相同，默认启用。由于至少一次投递，已配置时每次重启最多多出一次合并后的刷新。 + - **内置插件 `media-server-refresh`**：订阅 `torrent.organized`（另一份订阅 `file.renamed`；前者在重命名方式为 `none` 时也会发布），等待 `delay` 秒，把期间的事件合并成一次 Jellyfin / Emby / Plex 刷新请求。未填写地址或 API Key 时不做任何事，所以与另一份相同，默认启用。由于至少一次投递，已配置时每次重启最多多出一次合并后的刷新。它同时订阅 `hardlink` 在新放入文件后发布的 `hardlink.linked`（普通 `Event`，不是通知）：两者各自订阅同一个事件，跨盘复制慢于 `delay` 时刷新会早于文件到位，所以文件放好后再刷新一次；刷新请求发出后到达的事件再排一次刷新，不再被忽略。 - **死代码**：删除 `Renamer.rename_file` / `_rename_media_file`、`_lookup_offsets` / `_normalize_path`、`BangumiDatabase.match_by_save_path`、`TorrentDatabase.search_by_qb_hash`、`RenameOperationDatabase.release_replacement_lease`。删除前确认它们在本分支（含 `revision_saga.py`）没有生产调用方。原测试改为经 `rename()` 与 `_batch_lookup_offsets` 驱动。`season_offset` 从 `gen_path` 起整条重命名链路移除，`_batch_lookup_offsets` 的结果从 `(集数偏移, 季度偏移, 类型)` 改为 `(集数偏移, 类型)`：季度偏移已体现在 Season 文件夹，文件名从未使用它。 - **插件配置 422**：`field_validator` 抛出的 `ValueError` 会留在 `ValidationError.errors()` 的 `ctx` 中，无法 JSON 序列化，保存配置返回 500（`hardlink` 的路径校验、`rename` 的模板校验都会触发）。现在以 `include_context=False` 返回 422。 - **用户文档**：`docs/{,en/,ja/}config/manager.md` 增加 `template`、`hardlink`（含 `path_map` 与 Docker 下硬链接不能跨文件系统的说明）与 `media-server-refresh` 三节，按本分支的设计重写；`CHANGELOG.md` 增加 P4 条目。