fix(4.0-P3): 修复评审发现的插件过滤与返回值校验问题

- 手动收集整季与 eps_complete 补全（collect_season）同样经过 torrent.filter 钩子
- torrent.filter 收到的解析结果先经 title.parsed 钩子修正，与建规则时一致
- HookRunner 新增 check 校验：返回值校验在记成功之前，无效结果计入熔断
- torrent.adding 校验 save_path/category/tags 类型，无效时沿用原请求
- http.request 校验请求头为 str → str 映射，无效时沿用默认请求头
- metadata_provider 校验标题非空、季度为整数等字段，无效时不改写规则

backend/src/module/downloader/download_client.py

@@ -52,6 +52,16 @@ async def _fetch_torrent_files(
     return [f for group in groups for f in group]
 
 
+def _valid_add_request(request: AddRequest) -> bool:
+    """宿主直接使用这些字段；类型不对的结果按钩子失败处理（str 标签也拒绝）。"""
+    return (
+        isinstance(request.save_path, (str, type(None)))
+        and isinstance(request.category, str)
+        and isinstance(request.tags, (tuple, list))
+        and all(isinstance(t, str) for t in request.tags)
+    )
+
+
 async def _apply_adding_hooks(
     torrent: Torrent | list[Torrent],
     bangumi: Bangumi,
@@ -74,7 +84,9 @@ async def _apply_adding_hooks(
         category=category,
         tags=tuple(tags.split(",")) if tags else (),
     )
-    result = await runner.transform(points.TORRENT_ADDING, request, expect=AddRequest)
+    result = await runner.transform(
+        points.TORRENT_ADDING, request, expect=AddRequest, check=_valid_add_request
+    )
     new_tags = [t.strip() for t in result.tags if t.strip()]
     if tags and tags not in new_tags:
         logger.warning(

backend/src/module/manager/collector.py

@@ -55,6 +55,8 @@ class SeasonCollector:
                     link, bangumi.filter.replace(",", "|")
                 )
             torrents = [t for t in torrents if release_fits_bangumi(t.name, bangumi)]
+        # 手动收集整季与 eps_complete 补全同样经过 torrent.filter
+        torrents = [t for t in torrents if await RSSEngine.plugin_accepts(t, bangumi)]
         async with Database() as db:
             # bangumi 必须先落库拿到 id：add_torrent 用它打 ab:<id> 标签，
             # 种子行也要用它关联 bangumi_id——否则种子会被记成孤儿，

backend/src/module/network/request_url.py

@@ -1,6 +1,7 @@
 import asyncio
 import logging
 import math
+from collections.abc import Mapping
 from typing import Any
 
 import httpx
@@ -111,13 +112,22 @@ async def reset_shared_client():
     _shared_client_proxy_key = None
 
 
+def _valid_http_request(request: HttpRequest) -> bool:
+    """请求头须是 str → str 的映射，否则每个 GET 都会失败。"""
+    return isinstance(request.headers, Mapping) and all(
+        isinstance(k, str) and isinstance(v, str) for k, v in request.headers.items()
+    )
+
+
 async def _apply_http_hooks(url: str, headers: dict) -> dict:
     """``http.request``：插件按 URL 修改请求头（私有站 Cookie、UA 等）。"""
     runner = plugin_host.hook_runner(points.HTTP_REQUEST)
     if runner is None:
         return headers
     request = HttpRequest(method="GET", url=url, headers=dict(headers))
-    result = await runner.transform(points.HTTP_REQUEST, request, expect=HttpRequest)
+    result = await runner.transform(
+        points.HTTP_REQUEST, request, expect=HttpRequest, check=_valid_http_request
+    )
     # url / method 不允许被改写，只取请求头
     return dict(result.headers)
 

backend/src/module/plugin/runner.py

@@ -16,6 +16,8 @@ DEFAULT_HOOK_TIMEOUT = 30.0
 DEFAULT_FAILURE_THRESHOLD = 5
 
 TripCallback = Callable[[str, str], None]
+# 校验插件返回值（None 不经校验，表示不修改）
+Check = Callable[[Any], bool]
 
 
 class CircuitBreaker:
@@ -100,22 +102,24 @@ class HookRunner:
         value: Any,
         *args: Any,
         expect: type | tuple[type, ...] | None = None,
+        check: Check | None = None,
         **kwargs: Any,
     ) -> Any:
         """把 ``value`` 依次交给 transform 钩子；钩子失败或返回 None 时沿用原值。
 
-        ``expect`` 给出时，返回值不是该类型的结果按钩子失败处理（计入熔断），
-        避免一个插件的错误返回值传给后续钩子和宿主。
+        ``expect`` 给出时，返回值不是该类型的结果按钩子失败处理（计入熔断）；
+        ``check`` 再校验字段取值。避免一个插件的错误返回值传给后续钩子和宿主。
         """
+
+        def valid(result: Any) -> bool:
+            return (expect is None or isinstance(result, expect)) and (
+                check is None or check(result)
+            )
+
         for entry in self._registry.hooks(point, self._order(point)):
-            ok, result = await self._call(entry, point, (value, *args), kwargs)
+            ok, result = await self._call(entry, point, (value, *args), kwargs, valid)
             if not ok or result is None:
                 continue
-            if expect is not None and not isinstance(result, expect):
-                reason = f"{point} 返回了 {type(result).__name__}，已忽略"
-                logger.warning("[Plugin:%s] 钩子失败：%s", entry.plugin_id, reason)
-                self._breaker.record_failure(entry.plugin_id, reason)
-                continue
             value = result
         return value
 
@@ -126,13 +130,14 @@ class HookRunner:
         func: Callable[..., Any],
         *args: Any,
         timeout: float | None = None,
+        check: Check | None = None,
     ) -> tuple[bool, Any]:
         """以钩子同样的超时与熔断规则调用插件 Provider 的方法。
 
-        返回 ``(是否成功, 结果)``；失败已记录日志并计入熔断。
+        返回 ``(是否成功, 结果)``；失败（含 ``check`` 不通过）已记录日志并计入熔断。
         """
         entry = HookEntry(plugin_id, func, 0, timeout)
-        return await self._call(entry, point, args, {})
+        return await self._call(entry, point, args, {}, check)
 
     async def _call(
         self,
@@ -140,12 +145,16 @@ class HookRunner:
         point: str,
         args: tuple[Any, ...],
         kwargs: dict[str, Any],
+        check: Check | None = None,
     ) -> tuple[bool, Any]:
         timeout = entry.timeout or self._default_timeout
         try:
             result = entry.func(*args, **kwargs)
             if inspect.isawaitable(result):
                 result = await asyncio.wait_for(result, timeout)
+            # 返回值校验在记成功之前：无效结果必须计入熔断，而不是先清零再计一次
+            if result is not None and check is not None and not check(result):
+                raise TypeError(f"返回了无效结果 {type(result).__name__}，已忽略")
         except Exception as e:
             reason = (
                 f"{point} 超时（{timeout}s）"

backend/src/module/rss/engine.py

@@ -31,6 +31,7 @@ from module.parser.analyser.tokenizer import (
     ParsedRelease,
 )
 from module.parser.release_policy import preference_identity, preference_revision
+from module.parser.title_parser import _apply_title_hooks
 from module.plugin import host as plugin_host
 from module.plugin.views import bangumi_info, torrent_info
 
@@ -213,6 +214,9 @@ class RSSEngine:
             # 解析失败不应让过滤阶段中断整轮刷新，钩子收到 None
             logger.debug("Cannot parse %s for plugin filters: %s", torrent.name, e)
             release = None
+        if release is not None:
+            # 与建规则时一致：过滤钩子收到经 title.parsed 修正后的解析结果
+            release = await _apply_title_hooks(release)
         verdict = await runner.filter(
             points.TORRENT_FILTER,
             torrent_info(torrent),

backend/src/module/rss/metadata.py

@@ -66,6 +66,18 @@ CORE_PROVIDERS: dict[str, MetadataProvider] = {
 }
 
 
+def _valid_metadata(result: Metadata) -> bool:
+    """字段会直接写入新规则；标题为空、季度非整数等结果按失败处理。"""
+    return (
+        isinstance(result, Metadata)
+        and isinstance(result.official_title, str)
+        and result.official_title != ""
+        and isinstance(result.season, int)
+        and isinstance(result.year, (str, type(None)))
+        and isinstance(result.poster_link, (str, type(None)))
+    )
+
+
 async def _resolve(parser_id: str, request: MetadataRequest) -> Metadata | None:
     entry = (
         plugin_host.get_registry().providers(points.METADATA_PROVIDER).get(parser_id)
@@ -82,7 +94,7 @@ async def _resolve(parser_id: str, request: MetadataRequest) -> Metadata | None:
     runner = plugin_host.get_runner()
     if runner is not None:
         ok, result = await runner.call_provider(
-            entry.plugin_id, points.METADATA_PROVIDER, call
+            entry.plugin_id, points.METADATA_PROVIDER, call, check=_valid_metadata
         )
         if not ok:
             return None
@@ -92,9 +104,9 @@ async def _resolve(parser_id: str, request: MetadataRequest) -> Metadata | None:
         except Exception as e:
             logger.warning("[Plugin:%s] 元数据源失败：%s", entry.plugin_id, e)
             return None
-    if result is not None and not isinstance(result, Metadata):
+    if result is not None and not _valid_metadata(result):
         logger.warning(
-            "[Plugin:%s] 元数据源返回了 %s，已忽略",
+            "[Plugin:%s] 元数据源返回了无效结果 %s，已忽略",
             entry.plugin_id,
             type(result).__name__,
         )

backend/src/test/test_plugin_ingest.py

@@ -221,6 +221,47 @@ class TestTorrentFilter:
         sent = client.add_torrent.call_args.args[0]
         assert [t.url for t in sent] == ["https://e/11.torrent"]
 
+    async def test_collect_season_applies_filter(self, plugins):
+        from module.manager.collector import SeasonCollector
+
+        add_hook(
+            plugins.registry,
+            points.TORRENT_FILTER,
+            lambda t, r, b: "- 12" not in t.name,
+        )
+        torrents = [
+            Torrent(name=TORRENT_NAME, url="https://e/12.torrent"),
+            Torrent(
+                name="[Sub] Mushoku Tensei - 11 [1080p].mkv", url="https://e/11.torrent"
+            ),
+        ]
+        req = AsyncMock()
+        req.get_torrents = AsyncMock(return_value=torrents)
+        client = AsyncMock()
+        client.add_torrent = AsyncMock(return_value=AddResult.ADDED)
+        bangumi = make_bangumi(title_raw="Mushoku Tensei", filter="")
+        with patch("module.manager.collector.RequestContent") as MockReq:
+            MockReq.return_value.__aenter__ = AsyncMock(return_value=req)
+            MockReq.return_value.__aexit__ = AsyncMock(return_value=False)
+            await SeasonCollector(client).collect_season(bangumi, "https://e/rss")
+        sent = client.add_torrent.call_args.args[0]
+        assert [t.url for t in sent] == ["https://e/11.torrent"]
+
+    async def test_filter_receives_title_parsed_release(self, rss_engine, plugins):
+        await seed(rss_engine)
+        add_hook(
+            plugins.registry, points.TITLE_PARSED, lambda r: replace(r, group="Alias")
+        )
+        add_hook(
+            plugins.registry,
+            points.TORRENT_FILTER,
+            lambda t, r, b: r is not None and r.group == "Alias",
+        )
+        client = await refresh_with(
+            rss_engine, [Torrent(name=TORRENT_NAME, url="https://e/1.torrent")]
+        )
+        client.add_torrent.assert_called_once()
+
 
 # ---------------------------------------------------------------- title.parsed
 
@@ -346,6 +387,26 @@ class TestTorrentAdding:
         kwargs = mock_qb_client.add_torrents.call_args.kwargs
         assert (kwargs["category"], kwargs["tags"]) == ("Bangumi", "ab:3")
 
+    @pytest.mark.parametrize(
+        "changes",
+        [{"tags": None}, {"tags": "a,b"}, {"tags": (1,)}, {"save_path": 1}],
+    )
+    async def test_invalid_request_fields_fall_back_and_trip(
+        self, plugins, download_client, mock_qb_client, changes
+    ):
+        add_hook(
+            plugins.registry,
+            points.TORRENT_ADDING,
+            lambda request: replace(request, **changes),
+        )
+        bangumi = make_bangumi(id=3)
+        for _ in range(3):
+            assert await add_magnet(download_client, bangumi) is AddResult.ADDED
+        kwargs = mock_qb_client.add_torrents.call_args.kwargs
+        assert (kwargs["category"], kwargs["tags"]) == ("Bangumi", "ab:3")
+        assert kwargs["save_path"] == bangumi.save_path
+        assert plugins.tripped == ["ext"]
+
 
 # ---------------------------------------------------------------- http.request
 
@@ -384,6 +445,18 @@ class TestHttpRequest:
         headers = await self.fetch()
         assert "Cookie" not in headers and headers["User-Agent"]
 
+    @pytest.mark.parametrize("headers", [None, {"X-Retry": 1}])
+    async def test_invalid_headers_keep_defaults_and_trip(self, plugins, headers):
+        add_hook(
+            plugins.registry,
+            points.HTTP_REQUEST,
+            lambda request: replace(request, headers=headers),
+        )
+        for _ in range(3):
+            fetched = await self.fetch()
+        assert "X-Retry" not in fetched and fetched["User-Agent"]
+        assert plugins.tripped == ["ext"]
+
 
 # ---------------------------------------------------------------- metadata
 
@@ -445,6 +518,20 @@ class TestMetadataProvider:
         await RSSAnalyser().official_title_parser(bangumi, movie_rss, make_torrent())
         assert bangumi.official_title == "Frieren"
 
+    @pytest.mark.parametrize(
+        "changes", [{"official_title": None}, {"official_title": ""}, {"season": None}]
+    )
+    async def test_plugin_provider_invalid_fields_ignored(self, plugins, changes):
+        current = Metadata(official_title="Frieren", season=1)
+        add_metadata(plugins.registry, "bgm", FakeMetadata(replace(current, **changes)))
+        bangumi = make_bangumi(official_title="Frieren", season=1)
+        for _ in range(3):
+            await RSSAnalyser().official_title_parser(
+                bangumi, make_rss_item(parser="bgm"), make_torrent()
+            )
+        assert (bangumi.official_title, bangumi.season) == ("Frieren", 1)
+        assert plugins.tripped == ["ext"]
+
     async def test_unknown_parser_is_noop(self, plugins):
         bangumi = make_bangumi(official_title="A.B", poster_link="x")
         await RSSAnalyser().official_title_parser(