fix(search): restore concurrent progressive manual paging

This commit is contained in:
jxxghp
2026-10-08 11:57:10 +08:00
parent ed85f763ef
commit f0b34cf9f5
5 changed files with 204 additions and 32 deletions

View File

@@ -4,14 +4,15 @@ import asyncio
import base64
import binascii
import json
from contextlib import aclosing
from functools import partial
from typing import Any, AsyncIterator, Callable, Optional, TypeVar, cast
from typing import Any, AsyncGenerator, AsyncIterator, Callable, Optional, TypeVar, cast
from uuid import uuid4
from app.application.configuration import get_configured_system_config
from app.chain.media import MediaChain
from app.chain.search.contract import _SearchOwnerBase
from app.chain.search.execution import MediaSearchPlan
from app.chain.search.execution import MediaSearchPlan, _candidate_contexts
from app.chain.search.media import _build_missing_media_map
from app.chain.search.result import SearchResultOwner
from app.chain.search.scan import RESTART_AFTER_SECONDS, SearchSources
@@ -59,6 +60,7 @@ class _ManualPage:
"""一次单页请求只收集原始资源,结果处理委托给原搜索业务。"""
def __init__(self, owner: _SearchOwnerBase, plan: MediaSearchPlan, raw_title: bool, search_id: str) -> None:
"""持有本轮搜索计划和页摘要,候选只由事件编排层顺序收集。"""
self.owner, self.plan, self.raw_title = owner, plan, raw_title
self.media = cast(MediaInfo, plan.mediainfo)
self.sources = SearchSources(owner, plan)
@@ -66,7 +68,7 @@ class _ManualPage:
self.search_id = search_id
self.page_cache = Cache(ttl=RESTART_AFTER_SECONDS)
def read(self, key: str, page: int) -> dict[str, Any]:
def read(self, key: str, page: int) -> tuple[dict[str, Any], list[TorrentInfo]]:
"""上一页摘要保留 36 小时;失效时先补上一页再取目标页,只有目标页交给原结果处理。"""
query = self.sources.queries[key]
pageable = self.sources.pageable(key)
@@ -76,7 +78,7 @@ class _ManualPage:
"page": page, "can_continue": False, "error": None}
# 超过最大页数,或不支持分页的站点请求后续页时,直接按末页返回。
if page >= MAX_SEARCH_PAGES or (page > 0 and not pageable):
return facts
return facts, []
previous = self.page_cache.get(f"{source}:{page - 1}", region=_PAGE_CACHE_REGION) if page else None
pages = [page - 1, page] if page and previous is None else [page]
for requested_page in pages:
@@ -86,33 +88,61 @@ class _ManualPage:
if not (observation.attempted and observation.outcome == "success"):
facts.update(error=observation.error or "本页请求未完成,请继续加载重试",
can_continue=query["site"] != "plugin")
return facts
return facts, []
signature = search_page_signature(search_resource_id(item) for item in result.torrents) or ""
self.page_cache.set(f"{source}:{requested_page}", signature, region=_PAGE_CACHE_REGION)
if requested_page < page:
previous = signature
raw_count = observation.raw_count if observation.raw_count is not None else len(result.torrents)
if previous and signature == previous:
return facts
return facts, []
# 尚未确认末页时允许尝试下一页,不预取,也不保证下一页一定有结果。
facts.update(can_continue=bool(
pageable and raw_count and observation.has_more is not False and page + 1 < MAX_SEARCH_PAGES))
self.torrents.extend(result.torrents)
return facts
return facts, result.torrents
def results(self) -> list[Context]:
def results(self, torrents: Optional[list[TorrentInfo]] = None) -> list[Context]:
"""复用原搜索的识别、季集过滤和排序;显示过滤与分页能力无关。"""
torrents = self.torrents if torrents is None else torrents
if self.raw_title:
return SearchTitleOwner._resolve_title_request(cast(SearchTitleOwner, self.owner), _TitleSearchResolveRequest(
title=self.media.title, torrents=self.torrents, rule_groups=self.plan.rule_groups,
title=self.media.title, torrents=torrents, rule_groups=self.plan.rule_groups,
mtype=None if self.media.type == MediaType.UNKNOWN else self.media.type,
)).contexts
return SearchResultOwner._parse_result(
cast(SearchResultOwner, self.owner), torrents=self.torrents, mediainfo=self.media, keyword=self.plan.keyword,
cast(SearchResultOwner, self.owner), torrents=torrents, mediainfo=self.media, keyword=self.plan.keyword,
season_episodes=self.sources.season_episodes, rule_groups=self.plan.rule_groups,
custom_words=self.plan.custom_words, filter_params=self.plan.filter_params,
)
def preview(self, torrents: list[TorrentInfo]) -> list[Context]:
"""标题预览复用过滤,精确搜索先显示未匹配候选,最终整页统一校验。"""
return self.results(torrents) if self.raw_title else _candidate_contexts(self.media, torrents)
async def _read_pages(
manual: _ManualPage, keys: list[str], page: int,
) -> AsyncGenerator[tuple[str, dict[str, Any], list[TorrentInfo]], None]:
"""复用有界并发和取消收口;同站点多个关键词按轮次串行,先完成的页立即交付。"""
grouped: dict[str, list[str]] = {}
for key in keys:
grouped.setdefault(manual.sources.queries[key]["site"], []).append(key)
async def read_source(site: dict[str, Any], requested_page: int) -> list[Any]:
"""把指定来源的页事实适配到既有单页调度器。"""
return [await _offload(manual.read, site["key"], requested_page)]
for index in range(max(map(len, grouped.values()), default=0)):
pages = manual.owner._iter_site_page_results(
indexer_sites=[{"key": group[index]} for group in grouped.values() if index < len(group)],
search_pages=[page], search_page=read_source, should_continue=lambda _site, _items: False,
task_owner="chain.search.manual.site_page",
)
async with aclosing(pages):
async for site, _page, results, _continued, _elapsed in pages:
facts, torrents = results[0]
yield site["key"], facts, torrents
async def _prepare_page(owner: _SearchOwnerBase, params: dict[str, Any]) -> _ManualPage:
"""只构造本次请求的识别/关键词计划,不读写任何搜索检查点。"""
@@ -139,7 +169,7 @@ async def _prepare_page(owner: _SearchOwnerBase, params: dict[str, Any]) -> _Man
return await _offload(_ManualPage, owner, plan, raw_title, search_id)
async def _page_events(manual: _ManualPage, params: dict[str, Any]) -> AsyncIterator[dict[str, Any]]:
async def _page_events(manual: _ManualPage, params: dict[str, Any]) -> AsyncGenerator[dict[str, Any], None]:
"""每个来源只返回指定页;只有首页请求才使用别名回退和插件来源。"""
page = max(0, int(params.get("page") or 0))
initial = not params.get("source")
@@ -150,15 +180,27 @@ async def _page_events(manual: _ManualPage, params: dict[str, Any]) -> AsyncIter
keys = list(manual.sources.queries)
else:
keys = manual.sources.add_keywords(all_names=bool(manual.owner.runtime_config.search_multiple_name))
facts: list[dict[str, Any]] = []
facts_by_key: dict[str, dict[str, Any]] = {}
torrents_by_key: dict[str, list[TorrentInfo]] = {}
contexts: list[Context] = []
preview_count = 0
while keys:
for key in keys:
if key.startswith("plugin:") and not initial:
continue
facts.append(await _offload(manual.read, key, page))
yield {"type": "progress", "stage": "searching", "value": 90,
"text": "已请求一页,正在处理其他搜索来源"}
pages = _read_pages(manual, keys, page)
async with aclosing(pages):
async for key, facts, torrents in pages:
facts_by_key[key], torrents_by_key[key] = facts, torrents
manual.torrents.extend(torrents)
previews = await _offload(manual.preview, torrents)
preview_count += len(previews)
yield {"type": "append", "stage": "searching",
"value": len(facts_by_key) / len(manual.sources.queries) * 90,
"text": f"已完成 {len(facts_by_key)} / {len(manual.sources.queries)} 个搜索来源",
"items": [context.to_dict() for context in previews],
"total_items": preview_count, "candidate_items": len(manual.torrents)}
# 并发完成顺序只影响预览,完整结果仍保持原来源顺序和最终排序语义。
manual.torrents = [torrent for key in manual.sources.queries for torrent in torrents_by_key.get(key, [])]
yield {"type": "progress", "stage": "filtering", "value": 98,
"text": f"正在过滤匹配 {len(manual.torrents)} 个候选资源 ..."}
contexts = await _offload(manual.results)
if contexts or not initial:
break
@@ -173,9 +215,10 @@ async def _page_events(manual: _ManualPage, params: dict[str, Any]) -> AsyncIter
area=manual.plan.area, season=None if manual.raw_title else params.get("season"),
sites=params.get("sites"), result_type="torrent")
items = [context.to_dict() for context in contexts]
source_facts = [facts_by_key[key] for key in manual.sources.queries if key in facts_by_key]
yield {"type": "replace", "stage": "filtered", "value": 100, "items": items,
"total_items": len(items), "candidate_items": len(manual.torrents), "sources": facts}
yield {"type": "done", "stage": "done", "total_items": len(items), "sources": facts}
"total_items": len(items), "candidate_items": len(manual.torrents), "sources": source_facts}
yield {"type": "done", "stage": "done", "total_items": len(items), "sources": source_facts}
class SearchManualOwner:
@@ -189,5 +232,6 @@ class SearchManualOwner:
except ValueError as error:
yield {"type": "error", "message": str(error)}
return
async for event in _page_events(manual, params):
yield event
async with aclosing(_page_events(manual, params)) as events:
async for event in events:
yield event

View File

@@ -1174,4 +1174,4 @@ description、aliases、instructions,或通过 `append_instructions` 追加规
## Web 手动分页
`/api/v1/search/title/stream` 与 `/api/v1/search/media/{media_id}/stream` 新增可选参数 `manual_paging`、`page`、`source`。`manual_paging=true` 时每个来源只返回指定的一页,`replace`/`done` 事件附带各来源的 `source`、`site_name`、`page`、`can_continue`、`error`。不传该参数时行为不变,Agent `search.*` operation 不受影响。完整规则及订阅补全搜索策略 `SubscribeSearchStrategy` 见 [订阅补全搜索与手动分页](smart-search.md)。
`/api/v1/search/title/stream` 与 `/api/v1/search/media/{media_id}/stream` 新增可选参数 `manual_paging`、`page`、`source`。`manual_paging=true` 时每个来源只返回指定的一页,不同站点按既有并发数请求,并在来源完成后发送 `append` 预览;最终统一过滤后的 `replace`/`done` 事件附带各来源的 `source`、`site_name`、`page`、`can_continue`、`error`,预览不推进页号。不传该参数时行为不变,Agent `search.*` operation 不受影响。完整规则及订阅补全搜索策略 `SubscribeSearchStrategy` 见 [订阅补全搜索与手动分页](smart-search.md)。

View File

@@ -36,6 +36,7 @@
`/search/title/stream` 和 `/search/media/{id}/stream` 传 `manual_paging=true` 时启用单页模式:
- 首次请求(不带 `source`)每个来源请求第 0 页,只有这时才做别名回退和插件搜索。
- 不同站点按既有搜索并发数请求,同站点的多个关键词依次请求;每个来源完成后立即发送 `append` 预览,慢站点不会阻塞已完成来源的展示。标题预览遵循搜索过滤规则,精确媒体搜索先展示未匹配候选,最终结果统一匹配、过滤和排序。
- `replace` 和 `done` 事件带 `sources` 列表,每项包含不透明的 `source`、`site_name`、本次 `page`、`can_continue` 和 `error`。
- 续页时只回传 `source` 和要请求的 `page`(零基);客户端在整页结果应用后才推进页号,失败时保留页号以便重试。
- 后端按本轮来源和页号缓存原始页摘要 36 小时,用于判断与上一页是否重复;上一页摘要缺失时先补请求上一页。空页、重复页、明确末页、不支持分页的站点或达到 100 页时 `can_continue=false`。`can_continue=true` 只表示可以继续尝试,后端不预取、不自动重试。

View File

@@ -275,7 +275,9 @@ The `download.add` body must contain `torrent_in` (at least `title` and `enclosu
## Collection Counts And Pagination
- Search SSE `manual_paging=true` is a Web-only page-by-page mode and adds no Agent
operations; keep using the documented `search.*` operations.
operations; keep using the documented `search.*` operations. Web streams search
sites concurrently and emit `append` previews before the final filtered page;
only the completed result carries source state for advancing page numbers.
- For list inspection, explicitly send the operation's documented pagination
fields instead of requesting an unbounded legacy result. For optional legacy
pagination, start with `query={"page":1,"count":20}`.

View File

@@ -1,6 +1,10 @@
"""Web 手动分页:后端缓存页摘要并恢复过期比较依据,页号与重试由客户端持有。"""
import asyncio
import threading
from dataclasses import replace
from types import SimpleNamespace
from unittest.mock import AsyncMock
import pytest
@@ -8,17 +12,19 @@ from app.application.site.observation import report_site_search_outcome, report_
from app.chain.search import manual as manual_module
from app.chain.search.facade import SearchChain
from app.chain.search.manual import SearchManualOwner, decode_source, encode_source
from app.domain.context import TorrentInfo
from app.schemas.types import MediaType
from app.domain.context import MediaInfo, TorrentInfo
from app.schemas.types import MediaSource, MediaType
@pytest.fixture
def page_clock(monkeypatch):
"""使用可控时钟隔离页摘要缓存,验证失效后的上一页补取。"""
from app.runtime import cache as cache_module
clock = [0.0]
original = cache_module.MemoryTLRUCache.__init__
def initialize(cache, *args, **kwargs):
"""让测试缓存的过期判断统一使用可控时钟。"""
original(cache, *args, **kwargs, timer=lambda: clock[0])
monkeypatch.setattr(cache_module.MemoryTLRUCache, "__init__", initialize)
monkeypatch.setattr(cache_module.MemoryBackend, "_region_caches", {})
@@ -28,6 +34,7 @@ def page_clock(monkeypatch):
@pytest.fixture
def manual_owner(monkeypatch, page_clock):
"""构造离线搜索来源和保存边界,记录页请求并模拟失败。"""
owner = SearchChain()
sites = [{"id": 1, "name": "A"}, {"id": 2, "name": "B"}]
monkeypatch.setattr(owner, "_sync_indexers", lambda ids: [site for site in sites if not ids or site["id"] in ids])
@@ -36,11 +43,13 @@ def manual_owner(monkeypatch, page_clock):
saved = []
monkeypatch.setattr(owner, "save_last_search_params", lambda **params: saved.append(params))
async def save_results(contexts):
"""记录首次搜索的保存行为,避免写入真实缓存。"""
saved.append(len(contexts))
monkeypatch.setattr(owner, "_async_save_results", save_results)
calls = []
failures = set()
def request(**params):
"""按来源和页号构造结果,同时报告真实搜索观察字段。"""
site, page = params["site"]["id"], params["page"]
calls.append((site, params["keyword"], page))
failed = (site, page) in failures
@@ -53,13 +62,120 @@ def manual_owner(monkeypatch, page_clock):
async def events(iterator):
"""消费完整事件流,供最终页结果和来源状态断言使用。"""
return [item async for item in iterator]
def site_facts(result):
"""从最终状态中提取支持独立翻页的站点来源。"""
return [item for item in result[-1]["sources"] if decode_source(item["source"])[0] != "plugin"]
@pytest.fixture
def blocked_site(manual_owner, monkeypatch):
"""阻塞首个站点,其他站点即时完成,用同步信号验证并发和取消收口。"""
from app.runtime import tasks as tasks_module
owner, _, _ = manual_owner
monkeypatch.setattr(owner, "_runtime_config", replace(owner.runtime_config, search_threadpool_size=2))
monkeypatch.setattr(owner, "_runtime_config_provider", None)
registry = tasks_module.TaskRegistry()
monkeypatch.setattr(tasks_module, "_runtime_registry", registry)
started, release, finished = threading.Event(), threading.Event(), threading.Event()
original = owner.search_site_torrents
def request(**params):
"""让排序靠前的站点等待释放,证明快站点不受它阻塞。"""
if params["site"]["id"] != 1:
return original(**params)
started.set()
try:
assert release.wait(5), "测试未释放慢站点"
return original(**params)
finally:
finished.set()
monkeypatch.setattr(owner, "search_site_torrents", request)
yield started, release, finished, registry
release.set()
@pytest.mark.asyncio
@pytest.mark.parametrize("media_search", [False, True])
@pytest.mark.parametrize("transport", [False, True])
async def test_fast_site_preview_arrives_before_slow_site_and_final_page(
manual_owner, blocked_site, monkeypatch, media_search, transport,
):
"""标题和精确搜索均先展示快站点,最终过滤前不提交来源页号。"""
from app.chain.media import MediaChain
owner, calls, (_, saved) = manual_owner
started, release, finished, registry = blocked_site
params = {"keyword": "Show"}
if media_search:
media = MediaInfo(media_source=MediaSource.TMDB, media_id="1", tmdb_id=1, type=MediaType.TV,
title="Show", original_title="Show", names=["Show"])
monkeypatch.setattr(MediaChain, "async_recognize_media", AsyncMock(return_value=media))
monkeypatch.setattr(MediaChain, "async_supplement_media_info", AsyncMock(return_value=media))
params = {"media_source": MediaSource.TMDB, "media_id": "1", "mtype": MediaType.TV}
stream = SearchManualOwner.events(owner, params=params)
if transport:
from app.api.endpoints import search as search_endpoint
monkeypatch.setattr(search_endpoint, "_SSE_APPEND_FLUSH_INTERVAL", 0.01)
stream = search_endpoint._iter_batched_search_events(stream)
received = []
try:
async with asyncio.timeout(3):
while True:
event = await anext(stream)
received.append(event)
if event.get("items"):
break
assert started.is_set() and not finished.is_set()
assert event["type"] == "append" and event["stage"] == "searching"
assert [item["torrent_info"]["site"] for item in event["items"]] == [2]
assert "sources" not in event
assert not any(item["type"] in {"replace", "done"} for item in received)
assert saved == []
if media_search:
assert event["items"][0]["match_status"] == "candidate"
assert event["items"][0]["media_info_is_target"] is False
release.set()
received.extend(await events(stream))
finally:
release.set()
await stream.aclose()
assert sorted(calls) == [(1, "Show", 0), (2, "Show", 0)]
assert [item["type"] for item in received[-2:]] == ["replace", "done"]
assert len(received[-2]["items"]) == 2
assert [item["site_name"] for item in site_facts(received)] == ["A", "B"]
assert saved and not registry.records
@pytest.mark.asyncio
async def test_closing_preview_waits_for_inflight_workers_without_committing_page(manual_owner, blocked_site):
"""离开搜索流时等待在途同步请求收尾,不写最终结果和翻页状态。"""
owner, _, (_, saved) = manual_owner
started, release, finished, registry = blocked_site
stream = SearchManualOwner.events(owner, params={"keyword": "Show"})
try:
async with asyncio.timeout(3):
while not (await anext(stream)).get("items"):
pass
assert started.is_set() and not finished.is_set()
closing = asyncio.create_task(stream.aclose())
await asyncio.sleep(0)
assert not closing.done()
release.set()
await asyncio.wait_for(closing, timeout=3)
assert finished.is_set() and not registry.records
assert saved == []
finally:
release.set()
await stream.aclose()
@pytest.mark.asyncio
@pytest.mark.parametrize("requested,expected", [(None, None), (MediaType.UNKNOWN, None),
(MediaType.TV, MediaType.TV), (MediaType.MOVIE, MediaType.MOVIE)])
@@ -75,10 +191,14 @@ async def test_title_search_keeps_unrestricted_mteam_categories_and_explicit_typ
plugin_types = []
class FakeRequest:
"""回放 M-Team 的分类与分页响应,不访问真实站点。"""
def __init__(self, **_params):
"""接受原传输层参数以保持索引器调用协议。"""
pass
def post_res(self, _url=None, json=None, **_params):
"""用分类和页号生成可重复的离线搜索响应。"""
requests.append(json)
count = 1 if json["categories"] == MTorrentSpider._movie_category else 3
offset = (json["pageNumber"] - 1) * count
@@ -120,9 +240,10 @@ async def test_title_search_keeps_unrestricted_mteam_categories_and_explicit_typ
@pytest.mark.asyncio
async def test_first_request_loads_page_zero_of_every_source_and_saves_last_search(manual_owner):
"""首页并发请求各来源,但完成页事实仍保持来源顺序。"""
owner, calls, (_, saved) = manual_owner
result = await events(SearchManualOwner.events(owner, params={"keyword": "Show"}))
assert calls == [(1, "Show", 0), (2, "Show", 0)]
assert sorted(calls) == [(1, "Show", 0), (2, "Show", 0)]
sources = site_facts(result)
assert [(item["site_name"], item["page"], item["can_continue"], item["error"]) for item in sources] == [
("A", 0, True, None), ("B", 0, True, None)]
@@ -149,6 +270,7 @@ async def test_continuation_restores_previous_page_and_keeps_last_search(manual_
@pytest.mark.asyncio
@pytest.mark.parametrize("pageable", [True, False])
async def test_failed_source_keeps_its_page_for_retry_without_hiding_other_sources(manual_owner, monkeypatch, pageable):
"""来源失败不影响其他来源,重试只请求原失败页。"""
owner, calls, (failures, _) = manual_owner
monkeypatch.setattr(owner, "get_search_page_size", lambda **_params: 100 if pageable else None)
failures.add((1, 0))
@@ -156,11 +278,11 @@ async def test_failed_source_keeps_its_page_for_retry_without_hiding_other_sourc
first, second = site_facts(result)
assert (first["error"], first["page"], first["can_continue"]) == ("timeout", 0, True)
assert (second["error"], second["page"], second["can_continue"]) == (None, 0, pageable)
assert calls == [(1, "Show", 0), (2, "Show", 0)]
assert sorted(calls) == [(1, "Show", 0), (2, "Show", 0)]
failures.clear()
retried = await events(SearchManualOwner.events(owner, params={
"keyword": "Show", "page": first["page"], "source": first["source"]}))
assert calls == [(1, "Show", 0), (2, "Show", 0), (1, "Show", 0)]
assert sorted(calls) == [(1, "Show", 0), (1, "Show", 0), (2, "Show", 0)]
assert (site_facts(retried)[0]["error"], site_facts(retried)[0]["can_continue"]) == (None, pageable)
@@ -179,13 +301,15 @@ async def test_alias_fallback_only_on_first_page_and_source_keeps_actual_keyword
@pytest.mark.asyncio
async def test_zero_display_rows_still_have_more(manual_owner, monkeypatch):
"""标题预览和最终结果都为空时,原始页事实仍允许继续加载。"""
from app.chain.search.manual import _ManualPage
owner, _, _ = manual_owner
# 过滤规则把本页结果全部过滤掉:显示为空,但站点仍有后续页。
monkeypatch.setattr(_ManualPage, "results", lambda _self: [])
monkeypatch.setattr(_ManualPage, "results", lambda _self, _torrents=None: [])
result = await events(SearchManualOwner.events(owner, params={"keyword": "Show", "sites": [1]}))
assert result[-2]["items"] == []
assert all(not item["items"] for item in result if item["type"] == "append")
assert all(item["can_continue"] for item in site_facts(result))
@@ -231,6 +355,7 @@ async def test_site_without_pagination_has_only_the_first_page(manual_owner, mon
@pytest.mark.asyncio
@pytest.mark.parametrize("filter_all", [False, True])
async def test_raw_page_signature_stops_only_the_identical_adjacent_page(manual_owner, monkeypatch, filter_all):
"""原始页摘要独立于展示过滤,仅相邻页完全重复时结束来源。"""
from app.chain.search.manual import _ManualPage
owner, calls, _ = manual_owner
@@ -245,7 +370,7 @@ async def test_raw_page_signature_stops_only_the_identical_adjacent_page(manual_
for number in numbers]
monkeypatch.setattr(owner, "search_site_torrents", request)
if filter_all:
monkeypatch.setattr(_ManualPage, "results", lambda _self: [])
monkeypatch.setattr(_ManualPage, "results", lambda _self, _torrents=None: [])
first = await events(SearchManualOwner.events(owner, params={"keyword": "Show", "sites": [1]}))
first_source = site_facts(first)[0]
assert calls == [0]