nas/MoviePilot
1
0
Fork 0
mirror of https://github.com/jxxghp/MoviePilot.git synced 2026-11-03 22:04:57 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(search): load diagnostics lazily and cover user context forwarding

Browse Source
This commit is contained in:
SmallSpider
2026-11-01 10:07:09 +08:00
parent 9ce3d7a292
commit 05050376ea
9 changed files with 124 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
app/application/rules.py
View File

@@ -7,7 +7,7 @@ import copy
import threading
from collections.abc import Awaitable, Callable, Iterable, Mapping
from dataclasses import dataclass
from typing import Any, Dict, List, Optional, Protocol
from typing import TYPE_CHECKING, Any, Dict, List, Optional, Protocol
from typing import Literal as TypingLiteral
from pyparsing import (
@@ -37,9 +37,11 @@ from app.application.subscription.contract import (
from app.domain.context import MediaInfo
from app.schemas.common import JsonData
from app.schemas.rule import CustomRule, FilterRuleGroup
from app.schemas.search import SearchDiagnosticRuleGroup
from app.schemas.types import SystemConfigKey
if TYPE_CHECKING:
from app.schemas.search import SearchDiagnosticRuleGroup
_RULE_GROUP_LIST_CONFIG_KEYS = (
SystemConfigKey.SearchFilterRuleGroups,
SystemConfigKey.SubscribeFilterRuleGroups,
@@ -555,7 +557,7 @@ class RuleHelper:
def explain_rule_groups(
self, *, media: Optional[MediaInfo], group_names: list[str],
) -> list[SearchDiagnosticRuleGroup]:
) -> list["SearchDiagnosticRuleGroup"]:
"""
用现有规则选择策略说明所选规则的适用范围，不执行资源过滤。
@@ -565,6 +567,9 @@ class RuleHelper:
"""
if not group_names:
return []
# 搜索诊断在首次调用时加载，规则服务导入不扩大启动模块集合。
from app.schemas.search import SearchDiagnosticRuleGroup
configured = {group.name: group for group in self.get_rule_groups()}
effective = {group.name for group in self.get_rule_group_by_media(media=media, group_names=group_names)}
result = []

12
app/application/search/state.py
View File

@@ -2,12 +2,14 @@
搜索参数与结果缓存的应用服务。
"""
from typing import Any, Awaitable, Callable, Dict, List, Optional
from typing import TYPE_CHECKING, Any, Awaitable, Callable, Dict, List, Optional
from app.schemas.media import parse_media_key, resolve_media_identity
from app.schemas.search import SearchLastContextData
from app.schemas.types import MediaSource, MediaType
if TYPE_CHECKING:
from app.schemas.search import SearchLastContextData
def stringify_sites(sites: Optional[List[int]]) -> str:
"""
@@ -147,7 +149,7 @@ class SearchStateService:
"""
return normalize_search_params(await self._async_load_cache(self._params_key))
async def async_save_context(self, user_id: int, context: SearchLastContextData) -> None:
async def async_save_context(self, user_id: int, context: "SearchLastContextData") -> None:
"""
一次写入当前用户的完整参数、结果与诊断，避免并发请求交叉拼接。
@@ -156,13 +158,15 @@ class SearchStateService:
"""
await self._async_save_cache(context.model_dump(mode="json"), f"{self._params_key}_context_{user_id}")
async def async_load_context(self, user_id: int) -> SearchLastContextData:
async def async_load_context(self, user_id: int) -> "SearchLastContextData":
"""
只读取指定用户的完整快照，不回退到无法归属的全局散键。
:param user_id: 已由入口认证的用户主键
:return: 完整快照；首次搜索前返回空参数和结果且诊断为 None
"""
from app.schemas.search import SearchLastContextData
value = await self._async_load_cache(f"{self._params_key}_context_{user_id}")
return SearchLastContextData.model_validate(value) if isinstance(value, dict) else SearchLastContextData()

10
app/chain/search/cache.py
View File

@@ -2,7 +2,7 @@
搜索参数与结果缓存适配 owner。
"""
from typing import Any, Dict, List, Optional, cast
from typing import TYPE_CHECKING, Any, Dict, List, Optional, cast
from app.application.search.state import (
SearchStateService,
@@ -11,12 +11,14 @@ from app.application.search.state import (
)
from app.chain.search.contract import _SearchOwnerBase
from app.domain.context import Context, SubtitleInfo
from app.schemas.search import SearchLastContextData
from app.schemas.types import (
MediaSource,
MediaType,
)
if TYPE_CHECKING:
from app.schemas.search import SearchLastContextData
class SearchCacheOwner(_SearchOwnerBase):
"""
@@ -136,7 +138,7 @@ class SearchCacheOwner(_SearchOwnerBase):
"""
return await self._search_state().async_load_params()
async def async_last_search_context(self, user_id: int) -> SearchLastContextData:
async def async_last_search_context(self, user_id: int) -> "SearchLastContextData":
"""
读取已认证用户的原子搜索上下文，不与旧全局结果拼接。
@@ -145,7 +147,7 @@ class SearchCacheOwner(_SearchOwnerBase):
"""
return await self._search_state().async_load_context(user_id)
async def async_save_search_context(self, user_id: int, context: SearchLastContextData) -> None:
async def async_save_search_context(self, user_id: int, context: "SearchLastContextData") -> None:
"""
通过状态服务一次发布参数、结果和诊断。

10
app/chain/search/facade.py
View File

@@ -4,7 +4,7 @@
import threading
from collections.abc import AsyncIterator, Callable
from typing import Any, Optional, TypeVar, cast
from typing import TYPE_CHECKING, Any, Optional, TypeVar, cast
from app.application.subscription.sitebudget import (
SubscriptionSiteBudget,
@@ -27,9 +27,11 @@ from app.chain.search.title import SearchTitleOwner
from app.domain.context import Context, MediaInfo, MusicInfo, SubtitleInfo
from app.runtime.events import Event, eventmanager
from app.schemas.mediaserver import NotExistMediaInfo
from app.schemas.search import SearchLastContextData
from app.schemas.types import EventType, MediaSource, MediaType
if TYPE_CHECKING:
from app.schemas.search import SearchLastContextData
_PUBLIC_MODULE = "app.chain.search"
_Handler = TypeVar("_Handler", bound=Callable[..., Any])
@@ -199,7 +201,7 @@ class SearchChain(ChainBase):
save_last_search_params = SearchCacheOwner.save_last_search_params
async_save_last_search_params = SearchCacheOwner.async_save_last_search_params
async def async_last_search_context(self, user_id: int) -> SearchLastContextData:
async def async_last_search_context(self, user_id: int) -> "SearchLastContextData":
"""
读取当前认证用户的完整搜索快照，不读取全局散键。
@@ -208,7 +210,7 @@ class SearchChain(ChainBase):
"""
return await SearchCacheOwner.async_last_search_context(cast(SearchCacheOwner, self), user_id)
async def async_save_search_context(self, user_id: int, context: SearchLastContextData) -> None:
async def async_save_search_context(self, user_id: int, context: "SearchLastContextData") -> None:
"""
向状态 owner 委托发布当前用户的完整搜索快照。

36
app/chain/search/manual.py
View File

@@ -9,7 +9,7 @@ import json
from collections import Counter
from contextlib import aclosing
from functools import partial
from typing import Any, AsyncGenerator, AsyncIterator, Callable, Literal, Optional, TypeVar, cast
from typing import TYPE_CHECKING, Any, AsyncGenerator, AsyncIterator, Callable, Literal, Optional, TypeVar, cast
from uuid import uuid4
from app.application.configuration import get_configured_system_config
@@ -28,16 +28,11 @@ from app.domain.search import MAX_SEARCH_PAGES, search_page_signature, search_re
from app.runtime.cache import Cache
from app.runtime.execution import await_task_to_terminal
from app.runtime.tasks import get_task_registry
from app.schemas.search import (
SearchDiagnosticCounts,
SearchDiagnosticRejection,
SearchDiagnostics,
SearchDiagnosticSite,
SearchDiagnosticWarning,
SearchLastContextData,
)
from app.schemas.types import MediaType, SystemConfigKey
if TYPE_CHECKING:
from app.schemas.search import SearchDiagnostics, SearchDiagnosticSite
_PageResult = TypeVar("_PageResult")
_PAGE_CACHE_REGION = "search.manual.pages"
# 混合来源状态保留最需要用户注意的事实，不能把局部失败覆盖为空结果。
@@ -45,10 +40,12 @@ _OUTCOME_PRIORITY = {"empty": 0, "success": 1, "unknown": 2, "cooldown": 3, "err
def _site_diagnostic(site_id: str, site_name: Optional[str], observation: SiteSearchObservation,
returned: int) -> SearchDiagnosticSite:
returned: int) -> "SearchDiagnosticSite":
"""
仅投影可信的请求结果与数量，不复制可能含凭据的原始错误。
"""
from app.schemas.search import SearchDiagnosticSite
raw_count = observation.raw_count if observation.raw_count is not None else returned
outcome: Literal["success", "empty", "error", "cooldown", "unknown"] = "unknown"
if observation.outcome == "deferred":
@@ -114,9 +111,9 @@ class _ManualPage:
self.search_id = search_id
self.page_cache = Cache(ttl=RESTART_AFTER_SECONDS)
self.counts: Counter[str] = Counter()
self.observations: list[SearchDiagnosticSite] = []
self.observations: list["SearchDiagnosticSite"] = []
def read(self, key: str, page: int) -> tuple[dict[str, Any], list[TorrentInfo], list[SearchDiagnosticSite]]:
def read(self, key: str, page: int) -> tuple[dict[str, Any], list[TorrentInfo], list["SearchDiagnosticSite"]]:
"""
上一页摘要保留 36 小时；失效时先补上一页再取目标页，只有目标页交给原结果处理。
"""
@@ -126,7 +123,7 @@ class _ManualPage:
facts: dict[str, Any] = {"source": source,
"site_name": self.sources.sites.get(query["site"], {}).get("name"),
"page": page, "can_continue": False, "error": None}
observations: list[SearchDiagnosticSite] = []
observations: list["SearchDiagnosticSite"] = []
# 超过最大页数，或不支持分页的站点请求后续页时，直接按末页返回。
if page >= MAX_SEARCH_PAGES or (page > 0 and not pageable):
return facts, [], observations
@@ -173,13 +170,20 @@ class _ManualPage:
diagnostics=self.counts,
)
def diagnostics(self, contexts: list[Context]) -> SearchDiagnostics:
def diagnostics(self, contexts: list[Context]) -> "SearchDiagnostics":
"""
在完整页处理后生成请求独有的诊断，规则跳过不会伪装成资源拒绝。
:param contexts: 本页最终过滤结果
:return: 仅描述本次页请求的安全诊断快照
"""
from app.schemas.search import (
SearchDiagnosticCounts,
SearchDiagnosticRejection,
SearchDiagnostics,
SearchDiagnosticWarning,
)
rules = RuleHelper().explain_rule_groups(
media=None if self.raw_title else self.media, group_names=self.plan.rule_groups or [],
)
@@ -215,7 +219,7 @@ class _ManualPage:
async def _read_pages(
manual: _ManualPage, keys: list[str], page: int,
) -> AsyncGenerator[tuple[str, dict[str, Any], list[TorrentInfo], list[SearchDiagnosticSite]], None]:
) -> AsyncGenerator[tuple[str, dict[str, Any], list[TorrentInfo], list["SearchDiagnosticSite"]], None]:
"""
复用有界并发和取消收口；同站点多个关键词按轮次串行，先完成的页立即交付。
"""
@@ -319,6 +323,8 @@ async def _page_events(manual: _ManualPage, params: dict[str, Any]) -> AsyncGene
sites=params.get("sites"), result_type="torrent")
manual.owner.save_last_search_params(**cache_params)
if isinstance(params.get("user_id"), int):
from app.schemas.search import SearchLastContextData
await manual.owner.async_save_search_context(params["user_id"], SearchLastContextData.model_validate({
"params": SearchStateService.build_params(**cache_params) or {},
"results": [context.to_dict() for context in contexts], "diagnostics": diagnostics,

2
docs/architecture-overview.md
View File

@@ -821,7 +821,7 @@ flowchart LR
| 指标 | 当前值 |
|---|---:|
| Python 模块 | 1144 |
| 内部导入边 | 9,901 |
| 内部导入边 | 9,899 |
| 非平凡 SCC | 1（29 个模块的 TMDB vendor 精确 containment） |
| Application / Chain 具体 Adapter 直连 | 0 / 0 |
| Direct egress | 51（精确 policy 准入，Application / Chain 均为 0） |

6
tests/fixtures/architecture/dependency-baseline.json vendored
View File

@@ -959,8 +959,8 @@
"runtime_only": true
}
},
"edge_count": 9901,
"edge_sha256": "118c78bccde0ec5474ebdb81461411c538416b25a9a03e239647f925f9caf723",
"edge_count": 9899,
"edge_sha256": "fa4b6dd62edd439a931b77de55e0debf482fcc09c2caa223027a214959df9124",
"edges": [
"app -> app.foundation",
"app -> app.foundation.environment",
@@ -5269,7 +5269,6 @@
"app.chain.search.cache -> app.domain",
"app.chain.search.cache -> app.domain.context",
"app.chain.search.cache -> app.schemas",
"app.chain.search.cache -> app.schemas.search",
"app.chain.search.cache -> app.schemas.types",
"app.chain.search.contract -> app.chain",
"app.chain.search.contract -> app.chain.base",
@@ -5315,7 +5314,6 @@
"app.chain.search.facade -> app.runtime.events",
"app.chain.search.facade -> app.schemas",
"app.chain.search.facade -> app.schemas.mediaserver",
"app.chain.search.facade -> app.schemas.search",
"app.chain.search.facade -> app.schemas.types",
"app.chain.search.manual -> app.application",
"app.chain.search.manual -> app.application.configuration",

22
tests/test_search_architecture.py
View File

@@ -117,3 +117,25 @@ def test_search_facade_directly_extends_chain_base_with_stable_mro():
NotificationMixin,
object,
)
def test_search_diagnostic_models_stay_lazy_during_startup(tmp_path):
"""
共享 owner 不在应用冷导入时加载搜索诊断模型。
"""
script = """
import sys
from app.testing.bootstrap import ensure_sites_stub
ensure_sites_stub()
import app.factory
assert "app.schemas.search" not in sys.modules
"""
environment = os.environ.copy()
environment["CONFIG_DIR"] = str(tmp_path / "config")
result = subprocess.run(
[sys.executable, "-c", script], cwd=PROJECT_ROOT, env=environment,
check=False, capture_output=True, text=True, timeout=30,
)
assert result.returncode == 0, result.stderr or result.stdout

50
tests/test_search_diagnostics_api.py
View File

@@ -204,3 +204,53 @@ async def test_last_subtitle_context_signs_again_without_modifying_saved_snapsho
assert response.data.results[0].enclosure == "https://example.test/signed"
assert snapshot.results[0].enclosure == "https://example.test/subtitle"
assert signed[0][1]["purpose"] == endpoint.SecurityUtils.subtitle_download_purpose(1)
@pytest.mark.asyncio
async def test_real_search_chain_context_roundtrip_isolates_users_and_clears_diagnostics(monkeypatch):
"""
经真实门面、缓存 owner 与状态服务读写，用户快照保持独立且普通搜索清除旧诊断。
"""
from app.chain.search.facade import SearchChain
from app.domain.context import Context, TorrentInfo
chain = SearchChain()
stored = {}
async def save(value, key):
"""
用独立内存端口替代文件缓存，不执行外部 I/O。
"""
stored[key] = value
async def load(key):
"""
只返回请求用户对应的完整记录。
"""
return stored.get(key)
monkeypatch.setattr(chain, "async_save_cache", save)
monkeypatch.setattr(chain, "async_load_cache", load)
monkeypatch.setattr(endpoint, "SearchChain", lambda: chain)
snapshot = SearchLastContextData(params={"keyword": "First"}, diagnostics=SearchDiagnostics(
search_id="first-search", scope="page", stage_counts=SearchDiagnosticCounts(),
))
await chain.async_save_search_context(61, snapshot)
first = await endpoint.search_latest_context(TokenPayload(sub=61))
missing = await endpoint.search_latest_context(TokenPayload(sub=62))
assert first.data.params == {"keyword": "First"}
assert first.data.diagnostics.search_id == "first-search"
assert missing.data.model_dump() == {"params": {}, "results": [], "diagnostics": None}
await chain.async_save_search_context(62, SearchLastContextData(params={"keyword": "Second"}))
monkeypatch.setattr(chain, "async_search_by_title", AsyncMock(return_value=[
Context(torrent_info=TorrentInfo(title="Updated result")),
]))
await endpoint.search_by_title(keyword="Updated", _=TokenPayload(sub=61))
updated = await endpoint.search_latest_context(TokenPayload(sub=61))
second = await endpoint.search_latest_context(TokenPayload(sub=62))
assert updated.data.params["keyword"] == "Updated"
assert updated.data.diagnostics is None
assert updated.data.results[0].torrent_info.title == "Updated result"
assert second.data.params == {"keyword": "Second"}
assert second.data.results == []