fix(4.0): conflict_policy 与 media_files 插件经熔断器调用，失败时退回宿主实现

两者原先直接调用插件实现：抛出异常或返回值无效会中断整轮重命名，且不计入熔断。新增 plugin_host.call_sync（重命名策略同样改用它）， conflict_policy 失败按 hold 处理，media_files 失败按扩展名分类。
2026-11-06 07:29:09 +08:00 · 2026-10-08 16:34:20 +02:00
parent 04beea02c8
commit 1d764b0974
6 changed files with 135 additions and 36 deletions
--- a/backend/src/module/downloader/path.py
+++ b/backend/src/module/downloader/path.py
@@ -2,6 +2,7 @@ import logging
 import re
 from os import PathLike
 from pathlib import PureWindowsPath
+from typing import get_args

 from ab_sdk import points
 from ab_sdk.rename import CORE_ID, MediaFiles, MediaKind
@@ -24,6 +25,7 @@ else:

 _MEDIA_SUFFIXES = frozenset({".mp4", ".mkv"})
 _SUBTITLE_SUFFIXES = frozenset({".ass", ".srt"})
+_MEDIA_KINDS = get_args(MediaKind)

 # Windows/qB 保留字符 + 控制字符：出现在路径片段里会被下载器拆成多级
 # 目录或直接丢字（qB 静默截断），必须在拼路径前替换掉 (#721)
@@ -56,17 +58,24 @@ class SuffixMediaFiles:
 def check_files(files: list[dict]):
    # 选中的实现未登记（插件停用或被熔断）时退回宿主实现
    providers = plugin_host.get_registry().providers(points.MEDIA_FILES)
-    entry = providers.get(settings.plugins.slots.media_files) or providers[CORE_ID]
-    media_files: MediaFiles = entry.factory()
-    media_list = []
-    subtitle_list = []
-    for f in files:
-        file_name = f["name"]
-        kind = media_files.classify(file_name)
-        if kind == "media":
-            media_list.append(file_name)
-        elif kind == "subtitle":
-            subtitle_list.append(file_name)
+    core = providers[CORE_ID]
+    entry = providers.get(settings.plugins.slots.media_files) or core
+    names = [f["name"] for f in files]
+
+    def classify(media_files: MediaFiles) -> list[MediaKind]:
+        return [media_files.classify(name) for name in names]
+
+    ok, kinds = plugin_host.call_sync(
+        entry,
+        points.MEDIA_FILES,
+        classify,
+        lambda result: all(kind in _MEDIA_KINDS for kind in result),
+    )
+    if not ok:
+        # 插件抛出异常或返回了无效类别（已计入熔断）：本次按宿主实现分类
+        kinds = classify(core.factory())
+    media_list = [name for name, kind in zip(names, kinds) if kind == "media"]
+    subtitle_list = [name for name, kind in zip(names, kinds) if kind == "subtitle"]
    return media_list, subtitle_list


--- a/backend/src/module/manager/renamer.py
+++ b/backend/src/module/manager/renamer.py
@@ -7,7 +7,7 @@ from typing import Any
 from ab_sdk import points
 from ab_sdk.events import FileRenamed, OrganizedFile, TorrentOrganized
 from ab_sdk.rename import (
-    ConflictPolicy,
+    ConflictDecision,
    ConflictRequest,
    FileKind,
    RenameInput,
@@ -97,31 +97,17 @@ def _target_name(entry: ProviderEntry, f: RenameInput) -> str:
    插件策略经 runner 调用：异常与无效返回值计入熔断；策略主动抛出的
    RenameSkipped（输入或用户配置有问题）不计入。
    """
-    if entry.plugin_id == plugin_host.CORE:
-        return entry.factory().target_name(f)

-    def call() -> str | RenameSkipped:
+    def call(strategy: Any) -> str | RenameSkipped:
        try:
-            return entry.factory().target_name(f)
+            return strategy.target_name(f)
        except RenameSkipped as e:
            return e

-    runner = plugin_host.get_runner()
-    result: Any = None
-    if runner is not None:
-        ok, result = runner.call_provider_sync(
-            entry.plugin_id, points.RENAME_STRATEGY, call, check=_valid_name
-        )
-    else:
-        try:
-            result = call()
-            ok = _valid_name(result)
-        except Exception as e:
-            logger.warning("[Plugin:%s] 重命名策略失败：%s", entry.plugin_id, e)
-            ok = False
+    ok, result = plugin_host.call_sync(entry, points.RENAME_STRATEGY, call, _valid_name)
    if isinstance(result, RenameSkipped):
        raise result
-    if not ok or result is None:
+    if not ok:
        raise RenameSkipped(f"重命名方式 {entry.id} 执行失败，详见日志")
    return result

@@ -490,7 +476,7 @@ class Renamer(RevisionSaga):
        )
        if owners:
            owner = owners[0] if len(owners) == 1 else None
-            decision = self._conflict_policy().decide(
+            decision = self._decide_conflict(
                ConflictRequest(
                    target_path=prepared.target_path,
                    incoming=RevisionTask(
@@ -638,11 +624,19 @@ class Renamer(RevisionSaga):
        )

    @staticmethod
-    def _conflict_policy() -> ConflictPolicy:
-        # 选中的策略未登记（插件停用或被熔断）时按 hold 处理，不会误删旧版本
+    def _decide_conflict(request: ConflictRequest) -> ConflictDecision:
+        # 选中的策略未登记（插件停用或被熔断）、抛出异常或返回值无效时按 hold
+        # 处理，不会误删旧版本；插件的失败已计入熔断
        policies = plugin_host.get_registry().providers(points.CONFLICT_POLICY)
-        entry = policies.get(settings.plugins.slots.conflict_policy) or policies["hold"]
-        return entry.factory()
+        hold = policies["hold"]
+        entry = policies.get(settings.plugins.slots.conflict_policy) or hold
+        ok, decision = plugin_host.call_sync(
+            entry,
+            points.CONFLICT_POLICY,
+            lambda policy: policy.decide(request),
+            lambda result: isinstance(result, ConflictDecision),
+        )
+        return decision if ok else hold.factory().decide(request)

    async def _batch_lookup_offsets(
        self, torrents_info: list[dict]
--- a/backend/src/module/plugin/host.py
+++ b/backend/src/module/plugin/host.py
@@ -109,6 +109,36 @@ def hook_runner(point: str) -> HookRunner | None:
    return runner


+def call_sync(
+    entry: ProviderEntry,
+    point: str,
+    method: Callable[[Any], Any],
+    check: Callable[[Any], bool],
+) -> tuple[bool, Any]:
+    """同步调用 Provider 实现上的方法 ``method(impl)``，返回 ``(是否成功, 结果)``。
+
+    core 实现直接调用；插件实现经 runner 调用，异常与 ``check`` 不通过的返回值
+    计入熔断。未设置 runner（脚本、CLI）时插件失败只记录日志。``check`` 也会
+    收到 None。
+    """
+    if entry.plugin_id == CORE:
+        return True, method(entry.factory())
+    runner = _runner
+    if runner is not None:
+        ok, result = runner.call_provider_sync(
+            entry.plugin_id, point, lambda: method(entry.factory()), check=check
+        )
+    else:
+        try:
+            result = method(entry.factory())
+        except Exception as e:
+            logger.warning("[Plugin:%s] %s 失败：%s", entry.plugin_id, point, e)
+            return False, None
+        ok = True
+    # runner 不校验 None（钩子返回 None 表示不修改），Provider 返回 None 同样无效
+    return ok and check(result), result
+
+
 def provider_impls(
    point: str, valid: Callable[[Any], bool] = lambda impl: True
 ) -> dict[str, Any]:
--- a/backend/src/test/test_organize.py
+++ b/backend/src/test/test_organize.py
@@ -23,6 +23,7 @@ from ab_sdk.rename import (
 )
 from ab_sdk.testing import RecordingBus, create_plugin
 from module.downloader import DownloadClient
+from module.downloader.path import check_files
 from module.manager import renamer as renamer_module
 from module.manager.renamer import Renamer
 from module.manager.revision_policy import CoreConflictPolicy
@@ -226,6 +227,22 @@ async def test_rename_strategy_failure_keeps_name_and_notifies_once(
    assert plugins.tripped == (["ext"] if counted else [])


+@pytest.mark.parametrize("classify", [lambda path: 1 / 0, lambda path: "video"])
+def test_check_files_plugin_media_files_failure_falls_back_to_suffix(plugins, classify):
+    impl = SimpleNamespace(classify=classify)
+    plugins.registry.add_provider(
+        points.MEDIA_FILES, ProviderEntry("ext", "custom", lambda: impl)
+    )
+    files = [{"name": NAME}, {"name": SUB}, {"name": "readme.txt"}]
+
+    with patch("module.downloader.path.settings") as mock_settings:
+        mock_settings.plugins.slots.media_files = "custom"
+        results = [check_files(files), check_files(files)]
+
+    assert results == [([NAME], [SUB])] * 2
+    assert plugins.tripped == ["ext"]
+
+
 def _task(file_count: int = 1, revision: int | None = 1) -> RevisionTask:
    identity = (
        None
--- a/backend/src/test/test_renamer.py
+++ b/backend/src/test/test_renamer.py
@@ -1138,6 +1138,55 @@ class TestRevisionConflictFlow:
        assert restarted.events == []
        renamer.client.client.torrents_rename_file.assert_not_awaited()

+    @pytest.mark.parametrize(
+        "decide",
+        [
+            lambda request: 1 / 0,
+            lambda request: "replace",  # 不是 ConflictDecision
+        ],
+    )
+    async def test_plugin_conflict_policy_failure_holds_instead_of_aborting(
+        self, renamer, test_settings, decide
+    ):
+        from types import SimpleNamespace
+
+        from ab_sdk import points
+        from module.plugin import host
+        from module.plugin.registry import ProviderEntry
+
+        impl = SimpleNamespace(decide=decide)
+        host.get_registry().add_provider(
+            points.CONFLICT_POLICY, ProviderEntry("ext", "ext-policy", lambda: impl)
+        )
+        renamer.client.client.torrents_info.return_value = self._infos()
+
+        async def files(torrent_hash):
+            if torrent_hash == "old-v1":
+                return [{"name": self.TARGET}]
+            return [{"name": self.V2}]
+
+        renamer.client.client.torrents_files.side_effect = files
+        test_settings.plugins.slots.rename_strategy = "pn"
+        test_settings.plugins.slots.conflict_policy = "ext-policy"
+
+        try:
+            with (
+                patch("module.manager.renamer.settings", test_settings),
+                patch.object(
+                    renamer,
+                    "_batch_lookup_offsets",
+                    AsyncMock(return_value=self._offsets()),
+                ),
+            ):
+                assert await renamer.rename() == []
+        finally:
+            host.get_registry().remove_plugin("ext")
+
+        # 插件失败按 hold 处理：记录冲突并通知，不删除旧版本
+        assert len(renamer.events) == 1
+        renamer.client.client.torrents_rename_file.assert_not_awaited()
+        renamer.client.client.torrents_delete.assert_not_awaited()
+
    async def test_replace_stages_promotes_then_deletes_old(
        self, renamer, test_settings
    ):
--- a/docs/plans/2026-10-08-plugin-architecture-4.0-design.md
+++ b/docs/plans/2026-10-08-plugin-architecture-4.0-design.md
@@ -109,7 +109,7 @@

 slots 解析：

- `rename_strategy` 读 `slots.rename_strategy`，未登记时仍记录一次日志并按 `none` 处理。`media_files` 读 `slots.media_files`，未登记时退回 `default`。`conflict_policy` 读 `slots.conflict_policy`，未登记时退回 `hold`，不会误删旧版本。后两者不记日志，因为每个文件都会解析一次。
+- `rename_strategy` 读 `slots.rename_strategy`，未登记时仍记录一次日志并按 `none` 处理。`media_files` 读 `slots.media_files`，未登记时退回 `default`。`conflict_policy` 读 `slots.conflict_policy`，未登记时退回 `hold`，不会误删旧版本。后两者不记日志，因为每个文件都会解析一次。插件实现与重命名策略一样经 `plugin_host.call_sync`（runner 的熔断器）调用：抛出异常或返回值无效（`conflict_policy` 不是 `ConflictDecision`，`media_files` 不是三种类别之一）时计入熔断，本次按 `hold` / `default` 处理，不中断这一轮重命名。
 - **冲突策略改为两个 Provider**：第 9 节把 `revision_conflict_policy` 迁到 `slots.conflict_policy`，slot 的值就是 Provider id。因此宿主以 `core` 登记 `hold` 与 `replace` 两个 `CoreConflictPolicy`，不再登记 `default`；`ConflictRequest` 删除 P4 加入的 `configured` 字段。`ab_sdk` 升至 0.5.0。`GET /api/v1/plugins/providers` 仍不列出 `conflict_policy` / `media_files` 的插件候选。

 迁移器（第 9 节，`module/update/v4.py`）：