nas/Auto_Bangumi
1
0
Fork 0
mirror of https://github.com/EstrellaXD/Auto_Bangumi.git synced 2026-11-03 22:24:42 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(4.0-P4): 插件可通知事件进入通知中心，@subscribe 支持超时，清单增加 default_enabled

Browse Source 
- PluginBus 收到插件发布的 SystemEvent 时经 NotificationManager.send_event 投递（通知中心 → 事件总线 → 外部渠道）
- @subscribe(kind, timeout=...) 覆盖总线默认 30 秒的处理超时
- plugin.toml 的 default_enabled = false 让内置插件默认停用
This commit is contained in:
Estrella Pan
2026-10-08 10:51:33 +02:00
parent d645f1c8f6
commit 595d3746eb
6 changed files with 106 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
backend/src/ab_sdk/hooks.py
View File

@@ -32,6 +32,7 @@ class ProviderSpec:
@dataclass(frozen=True, slots=True)
class SubscribeSpec:
kind: str
timeout: float | None = None
@dataclass(frozen=True, slots=True)
@@ -83,15 +84,16 @@ def provider(point: str, *, id: str):
return mark
def subscribe(kind: str):
def subscribe(kind: str, *, timeout: float | None = None):
"""把方法声明为事件订阅者，参数为事件对象。
订阅者在独立队列中按发布顺序异步执行，失败或超时不影响发布方。
``kind`` 为 ``"*"`` 时接收全部事件。
``kind`` 为 ``"*"`` 时接收全部事件。``timeout`` 覆盖宿主默认的单个事件
处理超时（秒），适合复制大文件等耗时操作。
"""
def mark(func: F) -> F:
setattr(func, SUBSCRIBE_ATTR, SubscribeSpec(kind))
setattr(func, SUBSCRIBE_ATTR, SubscribeSpec(kind, timeout))
return func
return mark

9
backend/src/module/plugin/bus.py
View File

@@ -111,13 +111,18 @@ class EventBus:
self.on_error = on_error
def subscribe(
self, kind: str, handler: Handler, *, owner: str | None = None
self,
kind: str,
handler: Handler,
*,
owner: str | None = None,
timeout: float | None = None,
) -> Callable[[], None]:
sub = _Subscriber(
kind,
handler,
owner,
self._timeout,
timeout if timeout is not None else self._timeout,
self._queue_size,
self._report_error,
)

19
backend/src/module/plugin/context.py
View File

@@ -1,5 +1,6 @@
"""宿主侧的 PluginContext 实现。"""
import asyncio
import logging
from pathlib import Path
from typing import Any
@@ -7,11 +8,15 @@ from typing import Any
from pydantic import BaseModel
from ab_sdk import Event
from ab_sdk.events import SystemEvent
from .bus import EventBus, Handler
PLUGIN_DATA_ROOT = Path("config") / "plugin-data"
# 插件可通知事件的 fire-and-forget 发送任务，保留强引用防止被 GC
_notify_tasks: set[asyncio.Task] = set()
def _database():
# 延迟 import：module.plugin.host 被网络层、解析器等底层模块引用，
@@ -43,7 +48,11 @@ class DatabaseKV:
class PluginBus:
"""插件视角的总线：订阅归属到插件（卸载时统一清理、失败计入熔断），
发布的事件 kind 必须以 ``<plugin-id>.`` 开头，防止冒充宿主事件。"""
发布的事件 kind 必须以 ``<plugin-id>.`` 开头，防止冒充宿主事件。
插件发布的 :class:`SystemEvent` 与宿主事件走同一条通知路径
（``NotificationManager.send_event``：通知中心 → 事件总线 → 外部渠道）。
"""
def __init__(self, bus: EventBus, plugin_id: str) -> None:
self._bus = bus
@@ -55,6 +64,14 @@ class PluginBus:
f"插件 {self._plugin_id} 只能发布以 '{self._plugin_id}.' 开头的事件，"
f"收到 {event.kind!r}"
)
if isinstance(event, SystemEvent):
# 延迟 import：notification 依赖 module.plugin.host
from module.notification import NotificationManager
task = asyncio.create_task(NotificationManager().send_event(event))
_notify_tasks.add(task)
task.add_done_callback(_notify_tasks.discard)
return
self._bus.publish(event)
def subscribe(self, kind: str, handler: Handler):

12
backend/src/module/plugin/manager.py
View File

@@ -194,8 +194,9 @@ class PluginManager:
return None
def _enabled(self, candidate: PluginCandidate) -> bool:
"""启用开关；未设置时内置插件默认启用，其它来源默认禁用。"""
default = candidate.source == "builtin"
"""启用开关；未设置时内置插件默认启用（清单 ``default_enabled = false``
的除外），其它来源默认禁用。"""
default = candidate.source == "builtin" and candidate.manifest.default_enabled
return self._settings.plugins.enabled.get(candidate.manifest.id, default)
def _snapshot(self, plugin_id: str) -> str:
@@ -279,7 +280,12 @@ class PluginManager:
ProviderEntry(plugin_id, provider_spec.id, bound),
)
if subscribe_spec is not None:
self.bus.subscribe(subscribe_spec.kind, bound, owner=plugin_id)
self.bus.subscribe(
subscribe_spec.kind,
bound,
owner=plugin_id,
timeout=subscribe_spec.timeout,
)
async def _deactivate(self, plugin_id: str) -> None:
active = self._active.pop(plugin_id, None)

3
backend/src/module/plugin/manifest.py
View File

@@ -30,6 +30,9 @@ class PluginManifest(BaseModel):
description: str = ""
authors: list[str] = Field(default_factory=list)
permissions: list[str] = Field(default_factory=list)
default_enabled: bool = Field(
True, description="未设置启用开关时是否默认启用；只对内置插件生效"
)
@field_validator("id")
@classmethod

69
backend/src/test/test_plugin_runtime.py
View File

@@ -11,6 +11,7 @@ from typing import ClassVar, cast
import pytest
from ab_sdk import Event, Plugin, PluginDisabled, Verdict, hook, provider, subscribe
from ab_sdk.events import SystemEvent
from module.models.config import Plugins
from module.plugin import ExtensionPoint, ExtensionRegistry, RegistryError
from module.plugin.bus import EventBus
@@ -51,8 +52,11 @@ def write_plugin(
return plugin_dir
def manifest(plugin_id: str = "demo", sdk: str = ">=0.1,<1") -> PluginManifest:
return parse_manifest(MANIFEST.format(id=plugin_id, sdk=sdk, entry="x:Y"))
def manifest(
plugin_id: str = "demo", sdk: str = ">=0.1,<1", default_enabled: bool = True
) -> PluginManifest:
text = MANIFEST.format(id=plugin_id, sdk=sdk, entry="x:Y")
return parse_manifest(f"{text}default_enabled = {str(default_enabled).lower()}\n")
@dataclass(frozen=True, slots=True)
@@ -328,6 +332,33 @@ class TestHostContext:
ctx.bus.publish(PluginDisabled(plugin_id="x", reason="fake"))
await bus.close()
async def test_plugin_system_event_goes_through_notification_center(
self, tmp_path, monkeypatch
):
from module.notification.manager import NotificationManager
from module.plugin import context
@dataclass(frozen=True, slots=True)
class Synced(SystemEvent):
kind: ClassVar[str] = "demo.synced"
sent: list[Event] = []
async def send_event(self, event):
sent.append(event)
monkeypatch.setattr(NotificationManager, "send_event", send_event)
bus = EventBus()
on_bus: list[Event] = []
bus.subscribe("*", on_bus.append)
ctx = HostPluginContext("demo", None, bus, tmp_path)
ctx.bus.publish(Synced())
await asyncio.gather(*context._notify_tasks)
await bus.drain()
# send_event 自己负责发布到总线，PluginBus 不再重复入队
assert sent == [Synced()] and on_bus == []
await bus.close()
async def test_data_dir_created_lazily(self, tmp_path):
ctx = HostPluginContext("demo", None, EventBus(), tmp_path)
assert not (tmp_path / "demo").exists()
@@ -470,9 +501,12 @@ def candidate_for(
plugin_id: str = "demo",
source: str = "builtin",
sdk: str = ">=0.1,<1",
default_enabled: bool = True,
) -> PluginCandidate:
return PluginCandidate(
manifest=manifest(plugin_id, sdk), source=source, load=lambda: cls # type: ignore[arg-type]
manifest=manifest(plugin_id, sdk, default_enabled),
source=source, # type: ignore[arg-type]
load=lambda: cls,
)
@@ -558,10 +592,15 @@ class TestManager:
assert manager.statuses()[0].state == "active"
await manager.stop()
async def test_non_builtin_disabled_by_default(self, tmp_path):
@pytest.mark.parametrize(
"source, default_enabled", [("local", True), ("builtin", False)]
)
async def test_disabled_by_default(self, tmp_path, source, default_enabled):
recorder = Recorder()
manager, _ = make_manager(
candidate_for(build_plugin(recorder), source="local"),
candidate_for(
build_plugin(recorder), source=source, default_enabled=default_enabled
),
tmp_path=tmp_path,
allow_unsigned=True,
)
@@ -591,6 +630,26 @@ class TestManager:
assert len(seen) == 1
await manager.stop()
async def test_subscribe_timeout_overrides_bus_default(self, tmp_path):
seen: list[int] = []
class Slow(Plugin):
@subscribe("demo.happened", timeout=5)
async def on_demo(self, event):
await asyncio.sleep(0.05)
seen.append(event.value)
manager, _ = make_manager(candidate_for(Slow), tmp_path=tmp_path)
errors: list[str] = []
manager.bus = EventBus(
handler_timeout=0.01, on_error=lambda owner, reason: errors.append(reason)
)
await manager.start()
manager.bus.publish(DemoEvent(1))
await manager.bus.drain()
assert (seen, errors) == ([1], [])
await manager.stop()
async def test_failed_setup_rolls_back_registrations(self, tmp_path):
class Broken(Plugin):
@hook("torrent.filter")