From 58ed9a36b69b029c298613b074c165e49d9d3e46 Mon Sep 17 00:00:00 2001 From: Estrella Pan Date: Thu, 8 Oct 2026 09:19:03 +0200 Subject: [PATCH] =?UTF-8?q?fix(4.0-P2):=20=E6=A0=A1=E9=AA=8C=E6=8F=92?= =?UTF-8?q?=E4=BB=B6=E5=AE=9A=E6=97=B6=E4=BB=BB=E5=8A=A1=E9=97=B4=E9=9A=94?= =?UTF-8?q?=E4=B8=8E=E6=90=9C=E7=B4=A2=E7=AB=99=E7=82=B9/LLM=20=E9=80=82?= =?UTF-8?q?=E9=85=8D=E5=99=A8=E8=BF=94=E5=9B=9E=E5=80=BC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 定时任务的 interval/initial_delay 在受保护的 run() 中读取并校验，插件出错计入熔断，不再打断调度循环 - provider_impls 增加契约校验，搜索站点与 LLM 适配器返回值不合契约时跳过 --- backend/src/module/core/plugin_tasks.py | 29 +++++++++++++------ .../parser/analyser/providers/registry.py | 8 ++++- backend/src/module/plugin/host.py | 14 +++++++-- backend/src/module/searcher/provider.py | 10 ++++++- backend/src/test/test_plugin_extensions.py | 23 +++++++++++---- 5 files changed, 65 insertions(+), 19 deletions(-) diff --git a/backend/src/module/core/plugin_tasks.py b/backend/src/module/core/plugin_tasks.py index aac23d63..30b79f3d 100644 --- a/backend/src/module/core/plugin_tasks.py +++ b/backend/src/module/core/plugin_tasks.py @@ -47,9 +47,17 @@ class PluginTasks: plugin_id = entry.plugin_id try: spec: ScheduledTask = entry.factory() - spec_run, interval = spec.run, spec.interval - initial_delay = spec.initial_delay + spec_run, interval_src = spec.run, spec.interval + initial_delay = float(spec.initial_delay) enabled = spec.enabled or (lambda: True) + + def read_interval() -> float: + src = interval_src + return float(src() if callable(src) else src) + + # 间隔由插件代码给出，只在受保护的 run() 里读取；调度器拿到的是 + # 已校验的值，插件出错不会打断调度循环 + interval = read_interval() except Exception as e: reason = f"{points.SCHEDULED_TASK} {task_id}: {type(e).__name__}: {e}" logger.warning("[Plugin:%s] 定时任务创建失败：%s", plugin_id, reason) @@ -57,22 +65,25 @@ class PluginTasks: return None async def run() -> None: - # enabled 每轮重新读取（ScheduledTask 契约）；调度器只在启动时读 - # enabled，故任务总是启动，停用时空转跳过 + # enabled 与 interval 每轮重新读取（ScheduledTask 契约）；调度器只在 + # 启动时读 enabled，故任务总是启动，停用时空转跳过 + nonlocal interval try: - if not enabled(): - return - await spec_run() + ran = enabled() + if ran: + await spec_run() + interval = read_interval() except Exception as e: reason = f"定时任务 {task_id}: {type(e).__name__}: {e}" logger.warning("[Plugin:%s] %s", plugin_id, reason) self._breaker.record_failure(plugin_id, reason) return - self._breaker.record_success(plugin_id) + if ran: + self._breaker.record_success(plugin_id) return PeriodicTask( name=f"plugin:{plugin_id}:{task_id}", run=run, - interval=interval if callable(interval) else (lambda: float(interval)), + interval=lambda: interval, initial_delay=initial_delay, ) diff --git a/backend/src/module/parser/analyser/providers/registry.py b/backend/src/module/parser/analyser/providers/registry.py index bd020009..9f55d6f3 100644 --- a/backend/src/module/parser/analyser/providers/registry.py +++ b/backend/src/module/parser/analyser/providers/registry.py @@ -86,7 +86,13 @@ class ProviderRegistry: from module.plugin.host import provider_impls result: dict[str, type[LLMProviderAdapter]] = {} - for provider_id, adapter_cls in provider_impls(points.LLM_PROVIDER).items(): + + def valid(impl: object) -> bool: + return isinstance(impl, type) and issubclass(impl, LLMProviderAdapter) + + for provider_id, adapter_cls in provider_impls( + points.LLM_PROVIDER, valid + ).items(): if adapter_cls.info.id != provider_id: logger.warning( "LLM provider %s declares info.id=%s; skipped", diff --git a/backend/src/module/plugin/host.py b/backend/src/module/plugin/host.py index 6627c8fc..1cd74685 100644 --- a/backend/src/module/plugin/host.py +++ b/backend/src/module/plugin/host.py @@ -9,6 +9,7 @@ AB 自己的下载器、通知渠道与搜索站点和第三方插件走同一 """ import logging +from collections.abc import Callable from typing import Any from ab_sdk import points @@ -50,16 +51,23 @@ def provider(point: str, provider_id: str) -> Any | None: return entry.factory() if entry is not None else None -def provider_impls(point: str) -> dict[str, Any]: - """调用该扩展点全部 Provider 工厂；工厂出错的跳过并记录，不连累其它 Provider。""" +def provider_impls( + point: str, valid: Callable[[Any], bool] = lambda impl: True +) -> dict[str, Any]: + """调用该扩展点全部 Provider 工厂；工厂出错或产物不合契约（``valid`` + 返回 False）的跳过并记录，不连累其它 Provider。""" impls: dict[str, Any] = {} for provider_id, entry in get_registry().providers(point).items(): try: - impls[provider_id] = entry.factory() + impl = entry.factory() + if not valid(impl): + raise TypeError(f"不合契约的返回值 {impl!r}") except Exception as e: logger.warning( "[Plugin:%s] %s %s 创建失败：%s", entry.plugin_id, point, provider_id, e ) + continue + impls[provider_id] = impl return impls diff --git a/backend/src/module/searcher/provider.py b/backend/src/module/searcher/provider.py index 885a19c6..5219f28e 100644 --- a/backend/src/module/searcher/provider.py +++ b/backend/src/module/searcher/provider.py @@ -7,6 +7,14 @@ from module.models import RSSItem from module.plugin.host import provider_impls +def _valid_site(site: object) -> bool: + return ( + isinstance(site, SearchSite) + and isinstance(site.url, str) + and isinstance(site.parser, str) + ) + + def available_sites() -> dict[str, ProviderConfig]: """可用搜索站点：插件提供的站点 + 用户配置的站点（同名时用户配置优先）。 @@ -14,7 +22,7 @@ def available_sites() -> dict[str, ProviderConfig]: """ sites: dict[str, ProviderConfig] = {} site: SearchSite - for site_id, site in provider_impls(points.SEARCH_SITE).items(): + for site_id, site in provider_impls(points.SEARCH_SITE, _valid_site).items(): sites[site_id] = {"url": site.url, "parser": site.parser} sites.update(get_provider()) return sites diff --git a/backend/src/test/test_plugin_extensions.py b/backend/src/test/test_plugin_extensions.py index 992f0e3a..8b3082b2 100644 --- a/backend/src/test/test_plugin_extensions.py +++ b/backend/src/test/test_plugin_extensions.py @@ -207,14 +207,15 @@ class TestLLMProvider: with pytest.raises(ValueError): ProviderRegistry().resolve("claimed") - def test_provider_registry_broken_factory_skipped(self, registry): + @pytest.mark.parametrize("bad_factory", [broken_factory, object]) + def test_provider_registry_broken_factory_skipped(self, registry, bad_factory): from module.parser.analyser.providers.registry import ProviderRegistry add_plugin_provider( registry, points.LLM_PROVIDER, "ext-llm", make_adapter("ext-llm") ) registry.add_provider( - points.LLM_PROVIDER, ProviderEntry("bad", "bad-llm", broken_factory) + points.LLM_PROVIDER, ProviderEntry("bad", "bad-llm", bad_factory) ) llm_registry = ProviderRegistry() assert "ext-llm" in {i.id for i in llm_registry.list_infos()} @@ -253,14 +254,17 @@ class TestSearchSite: assert item.url == "https://acg/?q=Frieren+S2" assert item.parser == "tmdb" - def test_available_sites_broken_factory_skipped(self, registry): + @pytest.mark.parametrize( + "bad_factory", [broken_factory, object, lambda: SearchSite(None)] # type: ignore[arg-type] + ) + def test_available_sites_broken_factory_skipped(self, registry, bad_factory): from module.searcher import available_sites add_plugin_provider( registry, points.SEARCH_SITE, "acg", SearchSite("https://acg/?q=%s") ) registry.add_provider( - points.SEARCH_SITE, ProviderEntry("bad", "bad-site", broken_factory) + points.SEARCH_SITE, ProviderEntry("bad", "bad-site", bad_factory) ) with patch("module.searcher.provider.get_provider", return_value={}): sites = available_sites() @@ -344,7 +348,16 @@ class TestPluginTasks: await bridge.sync() assert scheduler.tasks == [] - @pytest.mark.parametrize("bad_spec", [None, object()]) + @pytest.mark.parametrize( + "bad_spec", + [ + None, + object(), + ScheduledTask(AsyncMock(), "abc"), # type: ignore[arg-type] + ScheduledTask(AsyncMock(), lambda: 1 / 0), + ScheduledTask(AsyncMock(), 60, initial_delay=None), # type: ignore[arg-type] + ], + ) async def test_sync_malformed_spec_skipped_and_recorded(self, registry, bad_spec): scheduler, bridge, tripped = self.make(registry, threshold=1) registry.add_provider(