From 05050376eac5241b2bd6c0ae0b0b40910d94197f Mon Sep 17 00:00:00 2001 From: SmallSpider <568442079@qq.com> Date: Sun, 1 Nov 2026 10:07:09 +0800 Subject: [PATCH] fix(search): load diagnostics lazily and cover user context forwarding --- app/application/rules.py | 11 ++-- app/application/search/state.py | 12 +++-- app/chain/search/cache.py | 10 ++-- app/chain/search/facade.py | 10 ++-- app/chain/search/manual.py | 36 +++++++------ docs/architecture-overview.md | 2 +- .../architecture/dependency-baseline.json | 6 +-- tests/test_search_architecture.py | 22 ++++++++ tests/test_search_diagnostics_api.py | 50 +++++++++++++++++++ 9 files changed, 124 insertions(+), 35 deletions(-) diff --git a/app/application/rules.py b/app/application/rules.py index 943d0ecdb..23b93656e 100644 --- a/app/application/rules.py +++ b/app/application/rules.py @@ -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 = [] diff --git a/app/application/search/state.py b/app/application/search/state.py index 6011f051f..812a943aa 100644 --- a/app/application/search/state.py +++ b/app/application/search/state.py @@ -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() diff --git a/app/chain/search/cache.py b/app/chain/search/cache.py index 53e111547..1d837d711 100644 --- a/app/chain/search/cache.py +++ b/app/chain/search/cache.py @@ -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: """ 通过状态服务一次发布参数、结果和诊断。 diff --git a/app/chain/search/facade.py b/app/chain/search/facade.py index 49c4211c3..46c87cab2 100644 --- a/app/chain/search/facade.py +++ b/app/chain/search/facade.py @@ -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 委托发布当前用户的完整搜索快照。 diff --git a/app/chain/search/manual.py b/app/chain/search/manual.py index 9a39218cc..a2b1ceef1 100644 --- a/app/chain/search/manual.py +++ b/app/chain/search/manual.py @@ -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, diff --git a/docs/architecture-overview.md b/docs/architecture-overview.md index 12a67b6e7..fd094d5ac 100644 --- a/docs/architecture-overview.md +++ b/docs/architecture-overview.md @@ -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） | diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index c939a23d3..cdb332c49 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -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", diff --git a/tests/test_search_architecture.py b/tests/test_search_architecture.py index 9a47fc143..939612dc3 100644 --- a/tests/test_search_architecture.py +++ b/tests/test_search_architecture.py @@ -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 diff --git a/tests/test_search_diagnostics_api.py b/tests/test_search_diagnostics_api.py index d905afded..2e1ccd4ba 100644 --- a/tests/test_search_diagnostics_api.py +++ b/tests/test_search_diagnostics_api.py @@ -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 == []