refactor(indexer): unify sync async search decisions

app/modules/indexer/__init__.py

@@ -1,12 +1,15 @@
+from dataclasses import dataclass
 from datetime import datetime
-from typing import List, Optional, Tuple, Union
+from types import MappingProxyType
+from typing import Any, Callable, List, Mapping, Optional, Tuple, Union, cast
 
-from app.domain.context import Context, SubtitleInfo, TorrentInfo
 from app.application.site.health import get_configured_site_health_service
 from app.application.site.query import get_configured_site_query_service
-from app.foundation.reflection import ModuleHelper
 from app.application.site.sites import SitesHelper  # pylint: disable=import-error,no-name-in-module
-from app.runtime.log import logger
+from app.domain import site as site_rules
+from app.domain.context import Context, SubtitleInfo, TorrentInfo
+from app.foundation import text as text_tools
+from app.foundation.reflection import ModuleHelper
 from app.modules import _ModuleBase
 from app.modules.indexer.parser import SiteParserBase
 from app.modules.indexer.spider import SiteSpider
@@ -17,13 +20,11 @@ from app.modules.indexer.spider.rousi import RousiSpider
 from app.modules.indexer.spider.sunnypt import SunnyPTSpider
 from app.modules.indexer.spider.tnode import TNodeSpider
 from app.modules.indexer.spider.torrentleech import TorrentLeech
-from app.schemas.types import MediaSource
-from app.schemas.media import resolve_media_identity
 from app.modules.indexer.spider.yema import YemaSpider
+from app.runtime.log import logger
+from app.schemas.media import resolve_media_identity
 from app.schemas.site import SiteUserData
-from app.schemas.types import MediaType, ModuleType, OtherModulesType
-from app.domain import site as site_rules
-from app.foundation import text as text_tools
+from app.schemas.types import MediaSource, MediaType, ModuleType, OtherModulesType
 
 SPIDER_PARSER_CLASSES = {
     "TNodeSpider": TNodeSpider,
@@ -36,6 +37,34 @@ SPIDER_PARSER_CLASSES = {
     "SunnyPT": SunnyPTSpider,
 }
 
+_SPECIALIZED_SEARCH_ARGUMENTS = {
+    "TNodeSpider": ("keyword", "page"),
+    "TorrentLeech": ("keyword", "mtype", "page"),
+    "mTorrent": ("keyword", "mtype", "page"),
+    "SunnyPT": ("keyword", "mtype", "cat", "page"),
+    "Yema": ("keyword", "mtype", "page"),
+    "Haidan": ("keyword", "mtype"),
+    "HDDolby": ("keyword", "mtype", "page"),
+    "RousiPro": ("keyword", "mtype", "cat", "page"),
+}
+
+
+@dataclass(frozen=True)
+class _IndexerSearchRequest:
+    """一次站点搜索的解析器选择与不可变调用参数"""
+
+    parser_class: Optional[Callable[[dict[str, Any]], Any]]
+    arguments: Mapping[str, Any]
+
+
+@dataclass(frozen=True)
+class _IndexerSearchOutcome:
+    """一次站点搜索完成后的错误状态、原始结果与耗时"""
+
+    error_flag: bool
+    result: List[dict[str, Any]]
+    seconds: int
+
 
 class IndexerModule(_ModuleBase):
     """
@@ -243,6 +272,107 @@ class IndexerModule(_ModuleBase):
             page_size = SiteSpider.default_result_num()
         return page_size if page_size > 0 else SiteSpider.default_result_num()
 
+    @staticmethod
+    def __create_search_request(
+        site: dict[str, Any],
+        keyword: Optional[str],
+        mtype: Optional[MediaType] = None,
+        cat: Optional[str] = None,
+        page: Optional[int] = 0,
+        search_type: str = "torrents",
+    ) -> Optional[_IndexerSearchRequest]:
+        """校验搜索条件并生成同步、异步共用的解析器调用请求"""
+        if search_type == "subtitles" and not site.get("subtitles"):
+            return None
+        if not IndexerModule.__search_check(site, keyword):
+            return None
+
+        search_word = IndexerModule.__clear_search_text(keyword)
+        if search_type == "subtitles":
+            return _IndexerSearchRequest(
+                parser_class=None,
+                arguments=MappingProxyType({
+                    "search_word": search_word,
+                    "indexer": site,
+                    "page": page,
+                    "search_type": search_type,
+                }),
+            )
+
+        parser_name = str(site.get("parser") or "")
+        parser_class = SPIDER_PARSER_CLASSES.get(parser_name)
+        argument_names = _SPECIALIZED_SEARCH_ARGUMENTS.get(parser_name)
+        if parser_class and argument_names:
+            available_arguments = {
+                "keyword": search_word,
+                "mtype": mtype,
+                "cat": cat,
+                "page": page,
+            }
+            return _IndexerSearchRequest(
+                parser_class=parser_class,
+                arguments=MappingProxyType({
+                    name: available_arguments[name]
+                    for name in argument_names
+                }),
+            )
+
+        return _IndexerSearchRequest(
+            parser_class=None,
+            arguments=MappingProxyType({
+                "search_word": search_word,
+                "indexer": site,
+                "mtype": mtype,
+                "cat": cat,
+                "page": page,
+            }),
+        )
+
+    @staticmethod
+    def __execute_search(
+        site: dict[str, Any],
+        request: _IndexerSearchRequest,
+    ) -> Tuple[bool, List[dict[str, Any]]]:
+        """通过同步解析器执行已冻结的搜索请求"""
+        if request.parser_class:
+            return cast(
+                Tuple[bool, List[dict[str, Any]]],
+                request.parser_class(site).search(**request.arguments),
+            )
+        return IndexerModule.__spider_search(**request.arguments)
+
+    @staticmethod
+    async def __async_execute_search(
+        site: dict[str, Any],
+        request: _IndexerSearchRequest,
+    ) -> Tuple[bool, List[dict[str, Any]]]:
+        """通过异步解析器执行已冻结的搜索请求"""
+        if request.parser_class:
+            return cast(
+                Tuple[bool, List[dict[str, Any]]],
+                await request.parser_class(site).async_search(**request.arguments),
+            )
+        return await IndexerModule.__async_spider_search(**request.arguments)
+
+    @staticmethod
+    def __create_search_outcome(
+        start_time: datetime,
+        error_flag: bool,
+        result: List[dict[str, Any]],
+    ) -> _IndexerSearchOutcome:
+        """把同步、异步 I/O 结果整理为共用的搜索完成状态"""
+        return _IndexerSearchOutcome(
+            error_flag=error_flag,
+            result=result,
+            seconds=(datetime.now() - start_time).seconds,
+        )
+
+    @staticmethod
+    def __log_search_error(site: dict[str, Any], search_type: str, error: Exception) -> None:
+        """按搜索类型记录同步、异步共用的操作失败信息"""
+        resource_name = "字幕" if search_type == "subtitles" else ""
+        logger.error(f"{site.get('name')} {resource_name}搜索出错：{str(error)}")
+
     def search_torrents(self, site: dict,
                         keyword: str = None,
                         mtype: MediaType = None,
@@ -265,85 +395,36 @@ class IndexerModule(_ModuleBase):
         # 错误标志
         error_flag = False
 
-        # 检查是否可以执行搜索
-        if not self.__search_check(site, keyword):
+        request = self.__create_search_request(
+            site=site,
+            keyword=keyword,
+            mtype=mtype,
+            cat=cat,
+            page=page,
+        )
+        if not request:
             return []
 
-        # 去除搜索关键字中的特殊字符
-        search_word = self.__clear_search_text(keyword)
-
         # 开始搜索
         try:
-            if site.get('parser') == "TNodeSpider":
-                error_flag, result = TNodeSpider(site).search(
-                    keyword=search_word,
-                    page=page
-                )
-            elif site.get('parser') == "TorrentLeech":
-                error_flag, result = TorrentLeech(site).search(
-                    keyword=search_word,
-                    mtype=mtype,
-                    page=page
-                )
-            elif site.get('parser') == "mTorrent":
-                error_flag, result = MTorrentSpider(site).search(
-                    keyword=search_word,
-                    mtype=mtype,
-                    page=page
-                )
-            elif site.get('parser') == "SunnyPT":
-                error_flag, result = SunnyPTSpider(site).search(
-                    keyword=search_word,
-                    mtype=mtype,
-                    cat=cat,
-                    page=page
-                )
-            elif site.get('parser') == "Yema":
-                error_flag, result = YemaSpider(site).search(
-                    keyword=search_word,
-                    mtype=mtype,
-                    page=page
-                )
-            elif site.get('parser') == "Haidan":
-                error_flag, result = HaiDanSpider(site).search(
-                    keyword=search_word,
-                    mtype=mtype
-                )
-            elif site.get('parser') == "HDDolby":
-                error_flag, result = HddolbySpider(site).search(
-                    keyword=search_word,
-                    mtype=mtype,
-                    page=page
-                )
-            elif site.get('parser') == "RousiPro":
-                error_flag, result = RousiSpider(site).search(
-                    keyword=search_word,
-                    mtype=mtype,
-                    cat=cat,
-                    page=page
-                )
-            else:
-                error_flag, result = self.__spider_search(
-                    search_word=search_word,
-                    indexer=site,
-                    mtype=mtype,
-                    cat=cat,
-                    page=page
-                )
+            error_flag, result = self.__execute_search(site, request)
         except Exception as err:
-            logger.error(f"{site.get('name')} 搜索出错：{str(err)}")
+            self.__log_search_error(site, "torrents", err)
 
-        # 索引花费的时间
-        seconds = (datetime.now() - start_time).seconds
+        outcome = self.__create_search_outcome(start_time, error_flag, result)
 
         # 统计索引情况
-        self.__indexer_statistic(site=site, error_flag=error_flag, seconds=seconds)
+        self.__indexer_statistic(
+            site=site,
+            error_flag=outcome.error_flag,
+            seconds=outcome.seconds,
+        )
 
         # 返回结果
         return self.__parse_result(
             site=site,
-            result_array=result,
-            seconds=seconds
+            result_array=outcome.result,
+            seconds=outcome.seconds,
         )
 
     def search_subtitles(self, site: dict,
@@ -361,30 +442,30 @@ class IndexerModule(_ModuleBase):
         start_time = datetime.now()
         error_flag = False
 
-        if not site.get("subtitles"):
+        request = self.__create_search_request(
+            site=site,
+            keyword=keyword,
+            page=page,
+            search_type="subtitles",
+        )
+        if not request:
             return []
 
-        if not self.__search_check(site, keyword):
-            return []
-
-        search_word = self.__clear_search_text(keyword)
-
         try:
-            error_flag, result = self.__spider_search(
-                search_word=search_word,
-                indexer=site,
-                page=page,
-                search_type="subtitles"
-            )
+            error_flag, result = self.__execute_search(site, request)
         except Exception as err:
-            logger.error(f"{site.get('name')} 字幕搜索出错：{str(err)}")
+            self.__log_search_error(site, "subtitles", err)
 
-        seconds = (datetime.now() - start_time).seconds
-        self.__indexer_statistic(site=site, error_flag=error_flag, seconds=seconds)
+        outcome = self.__create_search_outcome(start_time, error_flag, result)
+        self.__indexer_statistic(
+            site=site,
+            error_flag=outcome.error_flag,
+            seconds=outcome.seconds,
+        )
         return self.__parse_subtitle_result(
             site=site,
-            result_array=result,
-            seconds=seconds
+            result_array=outcome.result,
+            seconds=outcome.seconds,
         )
 
     async def async_search_torrents(self, site: dict,
@@ -409,85 +490,36 @@ class IndexerModule(_ModuleBase):
         # 错误标志
         error_flag = False
 
-        # 检查是否可以执行搜索
-        if not self.__search_check(site, keyword):
+        request = self.__create_search_request(
+            site=site,
+            keyword=keyword,
+            mtype=mtype,
+            cat=cat,
+            page=page,
+        )
+        if not request:
             return []
 
-        # 去除搜索关键字中的特殊字符
-        search_word = self.__clear_search_text(keyword)
-
         # 开始搜索
         try:
-            if site.get('parser') == "TNodeSpider":
-                error_flag, result = await TNodeSpider(site).async_search(
-                    keyword=search_word,
-                    page=page
-                )
-            elif site.get('parser') == "TorrentLeech":
-                error_flag, result = await TorrentLeech(site).async_search(
-                    keyword=search_word,
-                    mtype=mtype,
-                    page=page
-                )
-            elif site.get('parser') == "mTorrent":
-                error_flag, result = await MTorrentSpider(site).async_search(
-                    keyword=search_word,
-                    mtype=mtype,
-                    page=page
-                )
-            elif site.get('parser') == "SunnyPT":
-                error_flag, result = await SunnyPTSpider(site).async_search(
-                    keyword=search_word,
-                    mtype=mtype,
-                    cat=cat,
-                    page=page
-                )
-            elif site.get('parser') == "Yema":
-                error_flag, result = await YemaSpider(site).async_search(
-                    keyword=search_word,
-                    mtype=mtype,
-                    page=page
-                )
-            elif site.get('parser') == "Haidan":
-                error_flag, result = await HaiDanSpider(site).async_search(
-                    keyword=search_word,
-                    mtype=mtype
-                )
-            elif site.get('parser') == "HDDolby":
-                error_flag, result = await HddolbySpider(site).async_search(
-                    keyword=search_word,
-                    mtype=mtype,
-                    page=page
-                )
-            elif site.get('parser') == "RousiPro":
-                error_flag, result = await RousiSpider(site).async_search(
-                    keyword=search_word,
-                    mtype=mtype,
-                    cat=cat,
-                    page=page
-                )
-            else:
-                error_flag, result = await self.__async_spider_search(
-                    search_word=search_word,
-                    indexer=site,
-                    mtype=mtype,
-                    cat=cat,
-                    page=page
-                )
+            error_flag, result = await self.__async_execute_search(site, request)
         except Exception as err:
-            logger.error(f"{site.get('name')} 搜索出错：{str(err)}")
+            self.__log_search_error(site, "torrents", err)
 
-        # 索引花费的时间
-        seconds = (datetime.now() - start_time).seconds
+        outcome = self.__create_search_outcome(start_time, error_flag, result)
 
         # 统计索引情况
-        await self.__async_indexer_statistic(site=site, error_flag=error_flag, seconds=seconds)
+        await self.__async_indexer_statistic(
+            site=site,
+            error_flag=outcome.error_flag,
+            seconds=outcome.seconds,
+        )
 
         # 返回结果
         return self.__parse_result(
             site=site,
-            result_array=result,
-            seconds=seconds
+            result_array=outcome.result,
+            seconds=outcome.seconds,
         )
 
     async def async_search_subtitles(self, site: dict,
@@ -505,30 +537,30 @@ class IndexerModule(_ModuleBase):
         start_time = datetime.now()
         error_flag = False
 
-        if not site.get("subtitles"):
+        request = self.__create_search_request(
+            site=site,
+            keyword=keyword,
+            page=page,
+            search_type="subtitles",
+        )
+        if not request:
             return []
 
-        if not self.__search_check(site, keyword):
-            return []
-
-        search_word = self.__clear_search_text(keyword)
-
         try:
-            error_flag, result = await self.__async_spider_search(
-                search_word=search_word,
-                indexer=site,
-                page=page,
-                search_type="subtitles"
-            )
+            error_flag, result = await self.__async_execute_search(site, request)
         except Exception as err:
-            logger.error(f"{site.get('name')} 字幕搜索出错：{str(err)}")
+            self.__log_search_error(site, "subtitles", err)
 
-        seconds = (datetime.now() - start_time).seconds
-        await self.__async_indexer_statistic(site=site, error_flag=error_flag, seconds=seconds)
+        outcome = self.__create_search_outcome(start_time, error_flag, result)
+        await self.__async_indexer_statistic(
+            site=site,
+            error_flag=outcome.error_flag,
+            seconds=outcome.seconds,
+        )
         return self.__parse_subtitle_result(
             site=site,
-            result_array=result,
-            seconds=seconds
+            result_array=outcome.result,
+            seconds=outcome.seconds,
         )
 
     @staticmethod

docs/architecture-optimization-checklist.md

@@ -78,8 +78,8 @@ MoviePilot V3 已经形成较清晰的模块化单体：`foundation`、`domain`
 | Event Contract | 53 | 均已有 payload model，但当前全部是 diagnostic enforcement |
 | Python 源码量 | 约 298,700 行 | 60 个文件超过 1,000 行，10 个超过 2,000 行 |
 | 长方法 | 299 个超过 80 行 | 70 个超过 150 行，23 个超过 250 行；大量是私有方法 |
-| 全量 mypy 历史债务 | 10,245 / 592 文件 | 识别链 typed plan/result、TMDB 与消息双 ABI 共用计划继续清除参数、返回值和泛型注解债；strict frontier 当前覆盖 41 个文件 |
-| Ruff 历史诊断 | 642 | 低水位门禁通过，但规则集只覆盖 `E4/E7/E9/F/I` |
+| 全量 mypy 历史债务 | 10,233 / 592 文件 | 识别链 typed plan/result、TMDB、消息与 Indexer 双 ABI 共用计划继续清除参数、返回值和泛型注解债；strict frontier 当前覆盖 41 个文件 |
+| Ruff 历史诊断 | 641 | 低水位门禁通过，但规则集只覆盖 `E4/E7/E9/F/I` |
 | 覆盖率低水位 | Application 81.00%，Domain 80.67% | Chain、Runtime、Agent、Adapter、Startup 未进入包级覆盖率门禁 |
 
 ### 3.3 热点文件

tests/fixtures/architecture/mypy-baseline.json

@@ -2120,7 +2120,7 @@
     "type-arg": 18
   },
   "app/modules/indexer/__init__.py": {
-    "arg-type": 34,
+    "arg-type": 22,
     "assignment": 8,
     "attr-defined": 1,
     "empty-body": 1,

tests/fixtures/architecture/ruff-baseline.json

@@ -412,9 +412,6 @@
   "app/modules/filter/__init__.py": {
     "I001": 1
   },
-  "app/modules/indexer/__init__.py": {
-    "I001": 1
-  },
   "app/modules/indexer/parser/bitpt.py": {
     "I001": 1
   },

tests/test_indexer_sync_async_parity.py

@@ -0,0 +1,398 @@
+import asyncio
+
+import pytest
+
+import app.modules.indexer as indexer_module
+from app.modules.indexer import IndexerModule
+from app.schemas.types import MediaType
+
+
+def _install_search_observers(monkeypatch, events: list[tuple]) -> None:
+    """安装只记录业务顺序且不访问外部资源的搜索边界替身"""
+
+    def search_check(site, keyword=None):
+        events.append(("check", site["name"], keyword))
+        return True
+
+    def clear_search_text(keyword):
+        events.append(("normalize", keyword))
+        return "clean keyword"
+
+    def statistic(*, site, error_flag=False, seconds=0):
+        events.append(("statistic", site["name"], error_flag, seconds))
+
+    async def async_statistic(*, site, error_flag=False, seconds=0):
+        events.append(("statistic", site["name"], error_flag, seconds))
+
+    def parse_result(*, site, result_array, seconds):
+        events.append(("parse", site["name"], tuple(result_array), seconds))
+        return list(result_array)
+
+    def parse_subtitle_result(*, site, result_array, seconds):
+        events.append(("parse", site["name"], tuple(result_array), seconds))
+        return list(result_array)
+
+    monkeypatch.setattr(
+        IndexerModule,
+        "_IndexerModule__search_check",
+        staticmethod(search_check),
+    )
+    monkeypatch.setattr(
+        IndexerModule,
+        "_IndexerModule__clear_search_text",
+        staticmethod(clear_search_text),
+    )
+    monkeypatch.setattr(
+        IndexerModule,
+        "_IndexerModule__indexer_statistic",
+        staticmethod(statistic),
+    )
+    monkeypatch.setattr(
+        IndexerModule,
+        "_IndexerModule__async_indexer_statistic",
+        staticmethod(async_statistic),
+    )
+    monkeypatch.setattr(
+        IndexerModule,
+        "_IndexerModule__parse_result",
+        staticmethod(parse_result),
+    )
+    monkeypatch.setattr(
+        IndexerModule,
+        "_IndexerModule__parse_subtitle_result",
+        staticmethod(parse_subtitle_result),
+    )
+
+
+@pytest.mark.parametrize(
+    ("parser_name", "expected_arguments"),
+    [
+        ("TNodeSpider", {"keyword": "clean keyword", "page": 3}),
+        (
+            "TorrentLeech",
+            {"keyword": "clean keyword", "mtype": MediaType.MOVIE, "page": 3},
+        ),
+        (
+            "mTorrent",
+            {"keyword": "clean keyword", "mtype": MediaType.MOVIE, "page": 3},
+        ),
+        (
+            "SunnyPT",
+            {
+                "keyword": "clean keyword",
+                "mtype": MediaType.MOVIE,
+                "cat": "movie",
+                "page": 3,
+            },
+        ),
+        (
+            "Yema",
+            {"keyword": "clean keyword", "mtype": MediaType.MOVIE, "page": 3},
+        ),
+        (
+            "Haidan",
+            {"keyword": "clean keyword", "mtype": MediaType.MOVIE},
+        ),
+        (
+            "HDDolby",
+            {"keyword": "clean keyword", "mtype": MediaType.MOVIE, "page": 3},
+        ),
+        (
+            "RousiPro",
+            {
+                "keyword": "clean keyword",
+                "mtype": MediaType.MOVIE,
+                "cat": "movie",
+                "page": 3,
+            },
+        ),
+    ],
+)
+def test_torrent_search_uses_identical_parser_selection_and_arguments(
+    monkeypatch,
+    parser_name,
+    expected_arguments,
+):
+    """各专用解析器的同步、异步入口必须消费同一参数投影"""
+    events = []
+    calls = []
+    _install_search_observers(monkeypatch, events)
+
+    class FakeSpider:
+        """记录专用解析器收到的同步与异步调用"""
+
+        def __init__(self, site):
+            self.site = site
+
+        def search(self, **kwargs):
+            calls.append(("sync", self.site, kwargs))
+            events.append(("io", kwargs))
+            return False, [{"title": "hit"}]
+
+        async def async_search(self, **kwargs):
+            calls.append(("async", self.site, kwargs))
+            events.append(("io", kwargs))
+            return False, [{"title": "hit"}]
+
+    monkeypatch.setitem(indexer_module.SPIDER_PARSER_CLASSES, parser_name, FakeSpider)
+    site = {"id": 1, "name": "Parity", "parser": parser_name}
+    module = object.__new__(IndexerModule)
+
+    sync_result = module.search_torrents(
+        site=site,
+        keyword="Raw.Keyword",
+        mtype=MediaType.MOVIE,
+        cat="movie",
+        page=3,
+    )
+    sync_events = list(events)
+    events.clear()
+    async_result = asyncio.run(module.async_search_torrents(
+        site=site,
+        keyword="Raw.Keyword",
+        mtype=MediaType.MOVIE,
+        cat="movie",
+        page=3,
+    ))
+
+    assert sync_result == async_result == [{"title": "hit"}]
+    assert calls == [
+        ("sync", site, expected_arguments),
+        ("async", site, expected_arguments),
+    ]
+    assert sync_events == events
+    assert [event[0] for event in events] == [
+        "check",
+        "normalize",
+        "io",
+        "statistic",
+        "parse",
+    ]
+
+
+@pytest.mark.parametrize(
+    ("error_flag", "raw_result"),
+    [
+        (True, []),
+        (False, []),
+        (False, [{"title": "hit"}]),
+    ],
+)
+def test_torrent_search_preserves_result_and_status_parity(
+    monkeypatch,
+    error_flag,
+    raw_result,
+):
+    """成功、解析失败和空结果都必须以相同顺序完成统计与结果整理"""
+    events = []
+    _install_search_observers(monkeypatch, events)
+
+    class FakeSpider:
+        """返回参数化状态的专用解析器替身"""
+
+        def __init__(self, _site):
+            pass
+
+        def search(self, **_kwargs):
+            events.append(("io",))
+            return error_flag, list(raw_result)
+
+        async def async_search(self, **_kwargs):
+            events.append(("io",))
+            return error_flag, list(raw_result)
+
+    monkeypatch.setitem(indexer_module.SPIDER_PARSER_CLASSES, "TNodeSpider", FakeSpider)
+    site = {"id": 1, "name": "Parity", "parser": "TNodeSpider"}
+    module = object.__new__(IndexerModule)
+
+    sync_result = module.search_torrents(site=site, keyword="Raw.Keyword")
+    sync_events = list(events)
+    events.clear()
+    async_result = asyncio.run(module.async_search_torrents(site=site, keyword="Raw.Keyword"))
+
+    assert sync_result == async_result == raw_result
+    assert sync_events == events
+    assert events[-2][0:3] == ("statistic", "Parity", error_flag)
+    assert events[-1][0:3] == ("parse", "Parity", tuple(raw_result))
+
+
+def test_torrent_search_preserves_exception_fallback_parity(monkeypatch):
+    """解析器抛错后两种入口都沿用空结果与非解析错误统计语义"""
+    events = []
+    _install_search_observers(monkeypatch, events)
+
+    class FailingSpider:
+        """同步与异步均抛出相同操作错误的解析器替身"""
+
+        def __init__(self, _site):
+            pass
+
+        def search(self, **_kwargs):
+            events.append(("io",))
+            raise RuntimeError("broken")
+
+        async def async_search(self, **_kwargs):
+            events.append(("io",))
+            raise RuntimeError("broken")
+
+    monkeypatch.setitem(indexer_module.SPIDER_PARSER_CLASSES, "TNodeSpider", FailingSpider)
+    site = {"id": 1, "name": "Parity", "parser": "TNodeSpider"}
+    module = object.__new__(IndexerModule)
+
+    sync_result = module.search_torrents(site=site, keyword="Raw.Keyword")
+    sync_events = list(events)
+    events.clear()
+    async_result = asyncio.run(module.async_search_torrents(site=site, keyword="Raw.Keyword"))
+
+    assert sync_result == async_result == []
+    assert sync_events == events
+    assert events[-2][0:3] == ("statistic", "Parity", False)
+    assert events[-1][0:3] == ("parse", "Parity", ())
+
+
+def test_generic_torrent_search_uses_identical_request_projection(monkeypatch):
+    """普通页面 Spider 的同步、异步入口必须收到同一完整请求"""
+    events = []
+    calls = []
+    _install_search_observers(monkeypatch, events)
+
+    def spider_search(**kwargs):
+        calls.append(("sync", kwargs))
+        events.append(("io", kwargs))
+        return False, [{"title": "generic"}]
+
+    async def async_spider_search(**kwargs):
+        calls.append(("async", kwargs))
+        events.append(("io", kwargs))
+        return False, [{"title": "generic"}]
+
+    monkeypatch.setattr(
+        IndexerModule,
+        "_IndexerModule__spider_search",
+        staticmethod(spider_search),
+    )
+    monkeypatch.setattr(
+        IndexerModule,
+        "_IndexerModule__async_spider_search",
+        staticmethod(async_spider_search),
+    )
+    site = {"id": 1, "name": "Generic", "parser": "NexusPhp"}
+    module = object.__new__(IndexerModule)
+
+    sync_result = module.search_torrents(
+        site=site,
+        keyword="Raw.Keyword",
+        mtype=MediaType.TV,
+        cat="series",
+        page=2,
+    )
+    sync_events = list(events)
+    events.clear()
+    async_result = asyncio.run(module.async_search_torrents(
+        site=site,
+        keyword="Raw.Keyword",
+        mtype=MediaType.TV,
+        cat="series",
+        page=2,
+    ))
+
+    expected_arguments = {
+        "search_word": "clean keyword",
+        "indexer": site,
+        "mtype": MediaType.TV,
+        "cat": "series",
+        "page": 2,
+    }
+    assert sync_result == async_result == [{"title": "generic"}]
+    assert calls == [("sync", expected_arguments), ("async", expected_arguments)]
+    assert sync_events == events
+
+
+def test_rejected_torrent_search_skips_normalization_io_and_statistics(monkeypatch):
+    """搜索许可拒绝后同步、异步入口都必须立即停止后续工作"""
+    events = []
+
+    def search_check(_site, _keyword=None):
+        events.append(("check",))
+        return False
+
+    monkeypatch.setattr(
+        IndexerModule,
+        "_IndexerModule__search_check",
+        staticmethod(search_check),
+    )
+    site = {"id": 1, "name": "Rejected", "parser": "TNodeSpider"}
+    module = object.__new__(IndexerModule)
+
+    assert module.search_torrents(site=site, keyword="Raw.Keyword") == []
+    assert asyncio.run(module.async_search_torrents(site=site, keyword="Raw.Keyword")) == []
+    assert events == [("check",), ("check",)]
+
+
+def test_subtitle_search_uses_one_request_and_preserves_order(monkeypatch):
+    """字幕同步、异步入口必须共享许可、参数和状态整理顺序"""
+    events = []
+    calls = []
+    _install_search_observers(monkeypatch, events)
+
+    def spider_search(**kwargs):
+        calls.append(("sync", kwargs))
+        events.append(("io", kwargs))
+        return True, []
+
+    async def async_spider_search(**kwargs):
+        calls.append(("async", kwargs))
+        events.append(("io", kwargs))
+        return True, []
+
+    monkeypatch.setattr(
+        IndexerModule,
+        "_IndexerModule__spider_search",
+        staticmethod(spider_search),
+    )
+    monkeypatch.setattr(
+        IndexerModule,
+        "_IndexerModule__async_spider_search",
+        staticmethod(async_spider_search),
+    )
+    site = {"id": 1, "name": "Subtitle", "subtitles": {"search": {}}}
+    module = object.__new__(IndexerModule)
+
+    sync_result = module.search_subtitles(site=site, keyword="Raw.Keyword", page=4)
+    sync_events = list(events)
+    events.clear()
+    async_result = asyncio.run(module.async_search_subtitles(
+        site=site,
+        keyword="Raw.Keyword",
+        page=4,
+    ))
+
+    expected_arguments = {
+        "search_word": "clean keyword",
+        "indexer": site,
+        "page": 4,
+        "search_type": "subtitles",
+    }
+    assert sync_result == async_result == []
+    assert calls == [("sync", expected_arguments), ("async", expected_arguments)]
+    assert sync_events == events
+    assert [event[0] for event in events] == [
+        "check",
+        "normalize",
+        "io",
+        "statistic",
+        "parse",
+    ]
+    assert events[-2][0:3] == ("statistic", "Subtitle", True)
+
+
+def test_disabled_subtitle_search_skips_all_follow_up_work(monkeypatch):
+    """未启用字幕能力时两种入口都不得触发许可检查、I/O 或统计"""
+    events = []
+    _install_search_observers(monkeypatch, events)
+    site = {"id": 1, "name": "Subtitle", "subtitles": None}
+    module = object.__new__(IndexerModule)
+
+    assert module.search_subtitles(site=site, keyword="Raw.Keyword") == []
+    assert asyncio.run(module.async_search_subtitles(site=site, keyword="Raw.Keyword")) == []
+    assert events == []