diff --git a/CHANGELOG.md b/CHANGELOG.md index f4a9f0c6..baa8c854 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,7 +6,10 @@ - **新增（插件扩展点）**：插件可以提供下载器、通知渠道、LLM 解析提供商、搜索站点和定时任务（`ab_sdk.points`）。内置的 qB / aria2 和各通知渠道走同一套扩展注册表。设置页新增「插件」卡片，可查看状态、启停插件，并通过插件配置模型自动生成的表单修改配置；`secret_field` 声明的秘密字段在读取接口中一律掩码。下载器类型和通知渠道下拉框会列出插件提供的选项。新增插件开发文档（开发者指南 → 插件开发） - **新增（ingest 扩展点）**：RSS 下载流程开放 `torrent.filter`（种子过滤）、`title.parsed`（修正解析结果）、`torrent.adding`（修改保存路径 / 分类 / 标签）、`http.request`（为私有站点添加 Cookie 等请求头）四个钩子，以及 `metadata_provider` 元数据源扩展点：内置 mikan / tmdb 以 `core` 身份登记，插件可按 RSS 订阅的「解析器」取值提供新的元数据源，添加订阅时的解析器下拉框会列出插件选项。插件 SDK 升至 0.3.0，新增 `ab_sdk.ingest`。未安装插件时行为不变 - **新增（种子过滤）**：新增默认启用的内置插件「种子过滤」（`ingest-filters`）。在 设置 → 插件 中填写「包含过滤」后，只下载名称匹配任一表达式的种子（对所有订阅生效）；留空时不过滤 +- **新增（事件与外部接口）**：系统事件（RSS 订阅异常、种子添加失败、下载器不可用、新版本等）在写入通知中心后发布到插件事件总线，插件可用 `@subscribe("rss_failure")` 等订阅；事件类移入 `ab_sdk.events`（`SystemEvent` 基类，新增 `i18n()` 返回前端 i18n key 与参数）。插件可以提供 REST 路由（挂载于 `/api/v1/plugins//`，强制登录鉴权，随插件启停出现和消失）、MCP 工具（工具名 `<插件 id>__`）与 MCP 资源（`autobangumi://plugins/<插件 id>/`），并可通过 `message_template` 钩子按事件类型和渠道改写系统通知的标题与正文。`ab_sdk` 升至 0.3.0 +- **优化（通知中心推送）**：SSE 的 `notification` 帧改为订阅事件总线，通知中心有新消息、已读或删除时立即推送（原先每 3 秒比较一次修订号），推送格式不变 - **修复（搜索）**：`GET /api/v1/search/provider` 读取的是导入时的站点快照，新保存的搜索站点要重启后才会出现在列表中；现在实时读取 +- **修复（MCP）**：MCP 客户端每次断开 SSE 连接后，日志都会出现 `AssertionError: Unexpected message http.response.start`（3.x 起存在）。原因是 `/sse` 端点在 SSE 响应结束后又返回了一个空 `Response`，现在改为裸 ASGI 端点 - **破坏性变更（升级路径）**：只支持从 3.3.x 升级。检测到更早版本的数据（`config/version.info` 低于 3.3，或残留 2.x 的 `data/data.json`）时拒绝启动，并提示先升级到最新 3.3.x 启动一次完成迁移 - **破坏性变更（API）**：移除 3.2 兼容的 GET 控制端点（`/api/v1/restart`、`/start`、`/stop`、`/shutdown`）和 `GET /api/v1/auth/refresh_token`，请改用 POST - **破坏性变更（配置）**：移除旧版 `experimental_openai` 配置节和通知的单 provider 旧字段（`type` / `token` / `chat_id`），3.3 已把它们迁移到 `llm` 与 `notification.providers`；移除废弃的 `normal` 重命名方式，已有配置自动改为语义相同的 `none` diff --git a/backend/src/ab_sdk/events.py b/backend/src/ab_sdk/events.py index b5ce02a2..2576ed99 100644 --- a/backend/src/ab_sdk/events.py +++ b/backend/src/ab_sdk/events.py @@ -27,3 +27,310 @@ class PluginDisabled(Event): kind: ClassVar[str] = "plugin.disabled" plugin_id: str reason: str + + +# --- P5：系统事件 ------------------------------------------------------------- +# +# 以下事件由宿主经 ``NotificationManager.send_event`` 发出：先写入站内通知中心， +# 再发布到事件总线（插件可 ``@subscribe(kind)``），最后推送到外部通知渠道。 +# ``kind`` 沿用 3.x 的取值（站内通知中心按 kind 存储与翻译），不带 ``.`` 前缀。 + + +@dataclass(frozen=True, slots=True) +class SystemEvent(Event): + """可通知事件的基类：同时服务外部推送、站内通知中心与事件总线。 + + - ``describe()`` 返回默认中文 (标题, 正文)，用于外部推送与通知中心兜底展示； + - ``i18n()`` 返回 (i18n key, 参数)，前端据此按当前语言渲染，key 形如 + ``notifications.kind.``，其下有 ``title`` / ``body`` 两条文案； + - ``severity`` 为 ``info`` / ``warning`` / ``error``； + - ``once=True`` 表示同 ``dedup_key()`` 终生只入库一次（如「新版本可用」）。 + + 插件也可以继承它定义自己的可通知事件（``kind`` 须以 ``.`` 开头）。 + """ + + kind: ClassVar[str] = "system" + severity: ClassVar[str] = "info" + once: ClassVar[bool] = False + + def dedup_key(self) -> str | None: + """通知中心的去重键；None 表示每次都新建一条。""" + return None + + def payload(self) -> dict: + """结构化字段，入库为 JSON，也是 i18n 文案的插值参数。""" + from dataclasses import fields + + return {f.name: getattr(self, f.name) for f in fields(self)} + + def describe(self) -> tuple[str, str]: + """默认 (标题, 正文)。""" + return (self.kind, "") + + def i18n(self) -> tuple[str, dict]: + """(i18n key, 插值参数)。默认 key 为 ``notifications.kind.``。""" + return f"notifications.kind.{self.kind}", self.payload() + + +@dataclass(frozen=True, slots=True) +class RssFailureEvent(SystemEvent): + """RSS 订阅连接状态从正常变为异常（仅状态翻转时触发一次，而非每个 tick）。""" + + kind: ClassVar[str] = "rss_failure" + severity: ClassVar[str] = "error" + once: ClassVar[bool] = False + + rss_name: str + rss_url: str + error: str + + def dedup_key(self) -> str | None: + return f"rss_failure:{self.rss_url}" + + def payload(self) -> dict: + return {"rss_name": self.rss_name, "rss_url": self.rss_url, "error": self.error} + + def describe(self) -> tuple[str, str]: + """返回该事件的默认 (标题, 正文)。""" + return ( + "RSS 订阅连接异常", + f"订阅：{self.rss_name}\n地址：{self.rss_url}\n错误：{self.error}", + ) + + +@dataclass(frozen=True, slots=True) +class DownloadFailureEvent(SystemEvent): + """匹配到的种子在（内置 HTTP 重试后）仍添加下载器失败。""" + + kind: ClassVar[str] = "download_failure" + severity: ClassVar[str] = "error" + once: ClassVar[bool] = False + + official_title: str + torrent_name: str + + def dedup_key(self) -> str | None: + return f"download_failure:{self.official_title}" + + def payload(self) -> dict: + return { + "official_title": self.official_title, + "torrent_name": self.torrent_name, + } + + def describe(self) -> tuple[str, str]: + return ( + "种子添加失败", + f"番剧：{self.official_title}\n种子：{self.torrent_name}\n" + "重试后仍添加失败，请检查下载器连接。", + ) + + +@dataclass(frozen=True, slots=True) +class OffsetReviewEvent(SystemEvent): + """番剧被标记为需要人工确认季度/集数偏移。""" + + kind: ClassVar[str] = "offset_review" + severity: ClassVar[str] = "warning" + once: ClassVar[bool] = False + + official_title: str + reason: str + + def dedup_key(self) -> str | None: + return f"offset_review:{self.official_title}" + + def payload(self) -> dict: + return {"official_title": self.official_title, "reason": self.reason} + + def describe(self) -> tuple[str, str]: + return ( + "集数偏移待确认", + f"番剧：{self.official_title}\n原因：{self.reason}\n请前往设置页确认偏移量。", + ) + + +@dataclass(frozen=True, slots=True) +class DownloaderUnavailableEvent(SystemEvent): + """下载器不可用：连不上、用户名/密码错误或 IP 被封禁。 + + ``reason`` 只进 payload 不进 dedup_key——同一台下载器从 unreachable 翻转 + 到 credentials 时合并为一条（内容以最新为准），不产生两行。 + """ + + kind: ClassVar[str] = "downloader_unavailable" + severity: ClassVar[str] = "error" + once: ClassVar[bool] = False + + host: str + reason: str # unreachable | credentials | banned + + def dedup_key(self) -> str | None: + return f"downloader:{self.host}" + + def payload(self) -> dict: + return {"host": self.host, "reason": self.reason} + + def describe(self) -> tuple[str, str]: + details = { + "credentials": "用户名或密码错误，请在设置中检查下载器凭据。", + "banned": "IP 已被下载器封禁，请在下载器 WebUI 中解封或重启下载器。", + "unreachable": "无法连接下载器，请检查地址、端口和网络。", + } + detail = details.get(self.reason, details["unreachable"]) + return ("下载器连接异常", f"下载器：{self.host}\n{detail}") + + +@dataclass(frozen=True, slots=True) +class UpdateAvailableEvent(SystemEvent): + """检查到可用的新版本。""" + + kind: ClassVar[str] = "update_available" + severity: ClassVar[str] = "info" + once: ClassVar[bool] = True + + current: str + latest: str + channel: str + notes: str = "" + + def dedup_key(self) -> str | None: + return f"update_available:{self.latest}" + + def payload(self) -> dict: + return { + "current": self.current, + "latest": self.latest, + "channel": self.channel, + "notes": self.notes, + } + + def describe(self) -> tuple[str, str]: + return ( + "发现新版本", + f"当前版本：{self.current}\n最新版本：{self.latest}（{self.channel} 频道）\n" + "可前往 设置 → 软件更新 升级。", + ) + + +@dataclass(frozen=True, slots=True) +class UpdateAppliedEvent(SystemEvent): + """更新应用结果（成功或失败）。每次结果都单独入库，不做去重。""" + + once: ClassVar[bool] = False + + version: str + success: bool + message: str = "" + + # kind / severity 随结果变化，因此用属性覆盖基类的类变量 + @property + def kind(self) -> str: # type: ignore[override] + return "update_applied" if self.success else "update_failed" + + @property + def severity(self) -> str: # type: ignore[override] + return "info" if self.success else "error" + + def dedup_key(self) -> str | None: + return None + + def payload(self) -> dict: + return {"version": self.version, "message": self.message} + + def describe(self) -> tuple[str, str]: + if self.success: + return ("程序更新完成", f"已更新到 {self.version}，重启后生效。") + return ("程序更新失败", f"版本：{self.version}\n原因：{self.message}") + + +@dataclass(frozen=True, slots=True) +class LLMAuthFailureEvent(SystemEvent): + """订阅类 LLM 提供商凭据失效（刷新失败），需要用户重新连接。""" + + kind: ClassVar[str] = "llm_auth_failure" + severity: ClassVar[str] = "error" + once: ClassVar[bool] = False + + provider_id: str + account_label: str = "" + message: str = "" + + def dedup_key(self) -> str | None: + return f"llm_auth:{self.provider_id}" + + def payload(self) -> dict: + return { + "provider_id": self.provider_id, + "account_label": self.account_label, + "message": self.message, + } + + def describe(self) -> tuple[str, str]: + account = f"（{self.account_label}）" if self.account_label else "" + return ( + "LLM 提供商凭据失效", + f"提供商：{self.provider_id}{account}\n{self.message}\n" + "请前往 设置 → LLM 解析器 重新连接。", + ) + + +@dataclass(frozen=True, slots=True) +class LLMPluginInstallFailedEvent(SystemEvent): + """LLM 提供商插件安装失败（下载/签名校验/兼容性）。""" + + kind: ClassVar[str] = "llm_plugin_install_failed" + severity: ClassVar[str] = "error" + once: ClassVar[bool] = False + + plugin_id: str + version: str = "" + message: str = "" + + def dedup_key(self) -> str | None: + return f"llm_plugin_install:{self.plugin_id}" + + def payload(self) -> dict: + return { + "plugin_id": self.plugin_id, + "version": self.version, + "message": self.message, + } + + def describe(self) -> tuple[str, str]: + return ( + "LLM 插件安装失败", + f"插件：{self.plugin_id} {self.version}\n原因：{self.message}", + ) + + +@dataclass(frozen=True, slots=True) +class RenameConflictEvent(SystemEvent): + """媒体文件重命名遇到无法自动解决的目标路径冲突。""" + + kind: ClassVar[str] = "rename_conflict" + severity: ClassVar[str] = "warning" + once: ClassVar[bool] = False + + task_id: str + torrent_name: str + target_path: str + reason: str + + def dedup_key(self) -> str | None: + return f"rename_conflict:{self.task_id}:{self.target_path}" + + def payload(self) -> dict: + return { + "task_id": self.task_id, + "torrent_name": self.torrent_name, + "target_path": self.target_path, + "reason": self.reason, + } + + def describe(self) -> tuple[str, str]: + return ( + "媒体文件重命名冲突", + f"种子：{self.torrent_name}\n目标：{self.target_path}\n" + f"原因：{self.reason}", + ) diff --git a/backend/src/ab_sdk/mcp.py b/backend/src/ab_sdk/mcp.py new file mode 100644 index 00000000..77309976 --- /dev/null +++ b/backend/src/ab_sdk/mcp.py @@ -0,0 +1,52 @@ +"""MCP 工具与资源契约。 + +插件用 ``@provider(points.MCP_TOOL, id="search")`` 返回 :class:`McpTool`， +用 ``@provider(points.MCP_RESOURCE, id="stats")`` 返回 :class:`McpResource`。 +``id`` 只需在插件内唯一，宿主对外暴露时加上插件 id 前缀以避免冲突： + +- 工具名：``__``（双下划线。不用 ``.``，因为不少 MCP 客户端 + 所接的 LLM API 只接受 ``^[a-zA-Z0-9_-]{1,64}$`` 的工具名） +- 资源 URI：``autobangumi://plugins//`` + +MCP 端点（``/mcp``）沿用宿主的访问控制（IP 白名单 / ``scope=mcp`` 令牌）， +插件无需自行鉴权。 +""" + +from collections.abc import Awaitable, Callable +from dataclasses import dataclass, field +from typing import Any + +DEFAULT_TOOL_TIMEOUT = 60.0 + + +@dataclass(frozen=True) +class McpTool: + """一个 MCP 工具。 + + ``handler`` 接收客户端传入的参数字典（已按 ``input_schema`` 校验），返回 + 可 JSON 序列化的结果；宿主把结果序列化后作为文本内容返回给客户端。抛出的 + 异常会以 ``{"error": "..."}`` 返回，并计入插件熔断。 + """ + + description: str + handler: Callable[[dict[str, Any]], Awaitable[Any]] + # JSON Schema；默认不接受参数 + input_schema: dict[str, Any] = field( + default_factory=lambda: {"type": "object", "properties": {}} + ) + # 执行超时（秒） + timeout: float = DEFAULT_TOOL_TIMEOUT + + +@dataclass(frozen=True) +class McpResource: + """一个只读的 MCP 资源。 + + ``handler`` 不接受参数，返回字符串（原样返回）或可 JSON 序列化的对象 + （宿主序列化为 JSON）。 + """ + + name: str + handler: Callable[[], Awaitable[Any]] + description: str = "" + mime_type: str = "application/json" diff --git a/backend/src/ab_sdk/notify.py b/backend/src/ab_sdk/notify.py index a18bb795..89169189 100644 --- a/backend/src/ab_sdk/notify.py +++ b/backend/src/ab_sdk/notify.py @@ -46,3 +46,15 @@ class Notifier(Protocol): NotifierFactory = Callable[[NotifierSettings], Notifier] + + +@dataclass(frozen=True) +class RenderedMessage: + """``message_template`` 钩子处理的消息：系统事件推送前的标题与正文。 + + 初始值来自事件的 ``describe()``；钩子返回新的 RenderedMessage 即可改写， + 返回 None 表示不修改。多个钩子按优先级依次处理，后一个拿到前一个的结果。 + """ + + title: str + body: str diff --git a/backend/src/ab_sdk/points.py b/backend/src/ab_sdk/points.py index 315845ee..04552cee 100644 --- a/backend/src/ab_sdk/points.py +++ b/backend/src/ab_sdk/points.py @@ -32,3 +32,25 @@ TORRENT_FILTER = "torrent.filter" TITLE_PARSED = "title.parsed" TORRENT_ADDING = "torrent.adding" HTTP_REQUEST = "http.request" + +# --- P5：外部接口与通知模板 --- +# +# Provider 扩展点（id 只需在插件内唯一，宿主对外暴露时加插件 id 前缀）： +# +# - ``API_ROUTER``：返回 ``fastapi.APIRouter``，挂载于 +# ``/api/v1/plugins//``，强制登录鉴权 +# - ``MCP_TOOL``：返回 :class:`ab_sdk.mcp.McpTool`，MCP 工具名为 +# ``__`` +# - ``MCP_RESOURCE``：返回 :class:`ab_sdk.mcp.McpResource`，URI 为 +# ``autobangumi://plugins//`` +# +# Transform 钩子（``@hook``）： +# +# - ``MESSAGE_TEMPLATE``：签名 ``(message: RenderedMessage, event: SystemEvent, +# channel: str) -> RenderedMessage | None``，在系统事件推送到外部通知渠道前 +# 改写标题与正文；``channel`` 为渠道类型（如 ``telegram``） + +API_ROUTER = "api_router" +MCP_TOOL = "mcp_tool" +MCP_RESOURCE = "mcp_resource" +MESSAGE_TEMPLATE = "message_template" diff --git a/backend/src/module/api/__init__.py b/backend/src/module/api/__init__.py index db67cb5a..0fca0253 100644 --- a/backend/src/module/api/__init__.py +++ b/backend/src/module/api/__init__.py @@ -10,6 +10,7 @@ from .log import router as log_router from .movie import router as movie_router from .notification import router as notification_router from .passkey import router as passkey_router +from .plugin_routes import router as plugin_routes_router from .plugins import router as plugins_router from .program import router as program_router from .rss import router as rss_router @@ -41,3 +42,5 @@ v1.include_router(notification_router) v1.include_router(update_router) v1.include_router(llm_router) v1.include_router(plugins_router) +# 插件路由分发须排在插件管理路由之后（后者的路径更具体） +v1.include_router(plugin_routes_router) diff --git a/backend/src/module/api/events.py b/backend/src/module/api/events.py index 32fe0c12..4d84aa25 100644 --- a/backend/src/module/api/events.py +++ b/backend/src/module/api/events.py @@ -1,8 +1,13 @@ """SSE 端点：合并推送 status/downloader/log 更新，取代前端的三个轮询循环 (hooks/useAppInfo.ts:24、pages/index/downloader.vue:28、store/log.ts:25)。 -单个连接按 tick 节拍分别推送三种事件，节拍与原轮询间隔保持一致， -前端订阅失败时回退到原有轮询逻辑。 +单个连接按 tick 节拍分别推送 status/downloader/log/update，节拍与原轮询间隔 +保持一致，前端订阅失败时回退到原有轮询逻辑。这几类是「状态快照」而非事件， +仍按节拍采样。 + +notification（通知中心未读数）由事件驱动：每个连接在插件事件总线上订阅 +``inbox.changed``（通知中心写入/已读/删除时发布），收到后立即查库推送， +不再按节拍比较修订号。连接建立时推送一帧作为初始状态。 """ import asyncio @@ -18,7 +23,8 @@ from module.conf import LOG_PATH, VERSION from module.core import AppContext from module.database import Database from module.downloader import DownloadClient -from module.notification.inbox import inbox_revision +from module.notification.inbox import InboxChanged, inbox_revision +from module.plugin.host import get_bus from module.security.api import get_current_user from module.update import get_update_progress @@ -32,7 +38,6 @@ _TICK_SECONDS = 1 _STATUS_EVERY = 3 _DOWNLOADER_EVERY = 5 _LOG_EVERY = 10 -_NOTIFICATION_EVERY = 3 # 下载器查询的超时上限（秒）。下载器不可达时 qB/aria2 的认证重试可能阻塞 # 20-30s，而 SSE 是单连接串行推送——不设上限会把 status/log 事件一起卡住。 @@ -87,57 +92,112 @@ async def _log_payload() -> str | None: return data.decode("utf-8", errors="replace") +class _InboxWatch: + """一个 SSE 连接对通知中心变化的订阅。 + + 总线回调只置位 ``wake``；``pending`` 表示还欠前端一帧（初始帧、或上次查库 + 失败需要重试）。帧的 ``revision`` 取推送时的修订号，与旧版轮询一致。 + """ + + def __init__(self) -> None: + self.wake = asyncio.Event() + self.pending = True + bus = get_bus() + self._unsubscribe = ( + bus.subscribe(InboxChanged.kind, self._on_change) + if bus is not None + else None + ) + + def _on_change(self, _event) -> None: + self.wake.set() + + def close(self) -> None: + if self._unsubscribe is not None: + self._unsubscribe() + self._unsubscribe = None + + async def frame(self) -> dict | None: + """有待推送的变化时查库并返回 SSE 帧；没有或查库失败时返回 None。""" + if self.wake.is_set(): + self.wake.clear() + self.pending = True + if not self.pending: + return None + rev = inbox_revision() + try: + payload = await _notification_payload() + except Exception: + logger.debug("SSE: notification payload unavailable", exc_info=True) + return None + self.pending = False + payload["revision"] = rev + return {"event": "notification", "data": json.dumps(payload)} + + async def wait(self, deadline: float) -> bool: + """等到 ``deadline``（事件循环时间）或通知中心有变化；有变化返回 True。""" + remaining = deadline - asyncio.get_running_loop().time() + if remaining <= 0: + return False + try: + await asyncio.wait_for(self.wake.wait(), remaining) + except TimeoutError: + return False + return True + + async def _event_generator(request: Request, ctx: AppContext): - """按各自节拍推送 status/downloader/log 事件的核心循环。""" + """按各自节拍推送 status/downloader/log 事件、按需推送 notification 的核心循环。""" + loop = asyncio.get_running_loop() + inbox = _InboxWatch() tick = 0 last_update: str | None = None - last_inbox_rev: int | None = None - while True: - if await request.is_disconnected(): - break + try: + while True: + if await request.is_disconnected(): + break - # 更新进度：仅在有进行中的更新（phase != idle）且负载变化时推送， - # 避免重复帧刷屏；每 tick 检查一次以保证下载进度足够实时。 - progress = get_update_progress() - if progress.get("phase") != "idle": - payload = json.dumps(progress) - if payload != last_update: - last_update = payload - yield {"event": "update", "data": payload} + # 更新进度：仅在有进行中的更新（phase != idle）且负载变化时推送， + # 避免重复帧刷屏；每 tick 检查一次以保证下载进度足够实时。 + progress = get_update_progress() + if progress.get("phase") != "idle": + payload = json.dumps(progress) + if payload != last_update: + last_update = payload + yield {"event": "update", "data": payload} - if tick % _STATUS_EVERY == 0: - yield {"event": "status", "data": json.dumps(_status_payload(ctx))} + if tick % _STATUS_EVERY == 0: + yield {"event": "status", "data": json.dumps(_status_payload(ctx))} - if tick % _DOWNLOADER_EVERY == 0: - # 不可用时推送 null 而非跳过，前端据此感知下载器降级状态 - # （store/downloader.ts 对 null 保留旧数据，不会崩溃）。 - torrents = await _downloader_payload() - yield {"event": "downloader", "data": json.dumps(torrents)} + if tick % _DOWNLOADER_EVERY == 0: + # 不可用时推送 null 而非跳过，前端据此感知下载器降级状态 + # （store/downloader.ts 对 null 保留旧数据，不会崩溃）。 + torrents = await _downloader_payload() + yield {"event": "downloader", "data": json.dumps(torrents)} - if tick % _LOG_EVERY == 0: - log_text = await _log_payload() - if log_text is not None: - yield {"event": "log", "data": log_text} + if tick % _LOG_EVERY == 0: + log_text = await _log_payload() + if log_text is not None: + yield {"event": "log", "data": log_text} - # 通知中心：只在修订号变化时查库推送（写入/已读/删除都会 bump）， - # 平时每 3 tick 只做一次整数比较，不产生数据库查询。 - if tick % _NOTIFICATION_EVERY == 0: - rev = inbox_revision() - if rev != last_inbox_rev: - try: - notif_payload = await _notification_payload() - except Exception: - logger.debug("SSE: notification payload unavailable", exc_info=True) - else: - last_inbox_rev = rev - notif_payload["revision"] = rev - yield {"event": "notification", "data": json.dumps(notif_payload)} + # 通知中心：连接建立时的初始帧，以及此前查库失败后的重试 + frame = await inbox.frame() + if frame is not None: + yield frame - await asyncio.sleep(_TICK_SECONDS) - tick += _TICK_SECONDS + # 等待下一个 tick；其间 inbox.changed 到达时立即推送 + deadline = loop.time() + _TICK_SECONDS + while await inbox.wait(deadline): + frame = await inbox.frame() + if frame is not None: + yield frame + tick += 1 + finally: + inbox.close() @router.get("/stream", dependencies=[Depends(get_current_user)]) async def event_stream(request: Request, ctx: AppContext = Depends(get_context)): - """单一 SSE 连接，按各自节拍推送 status/downloader/log 事件。""" + """单一 SSE 连接，按各自节拍推送 status/downloader/log 事件， + 通知中心有变化时推送 notification 事件。""" return EventSourceResponse(_event_generator(request, ctx)) diff --git a/backend/src/module/api/plugin_routes.py b/backend/src/module/api/plugin_routes.py new file mode 100644 index 00000000..b4ce9b04 --- /dev/null +++ b/backend/src/module/api/plugin_routes.py @@ -0,0 +1,69 @@ +"""插件 REST 路由的分发端点：``/api/v1/plugins/{plugin_id}/{path:path}``。 + +鉴权在这里以路由依赖的形式强制（``get_current_user``），插件路由无法绕过， +也就无法注册匿名端点。请求转发给 ``AppContext.plugin_routes`` 中该插件的 +路由；插件未启用或未提供路由时返回 404。 + +该路由至少需要 ``/plugins//`` 之后的一段（可为空），因此不会遮蔽 +``module/api/plugins.py`` 中的 ``GET /plugins``、``PUT /plugins/settings``、 +``GET /plugins/providers`` 与 ``PUT /plugins/{id}``。 +""" + +from fastapi import APIRouter, Depends, HTTPException +from starlette.responses import Response +from starlette.types import Receive, Scope, Send + +from module.core import AppContext +from module.core.plugin_routes import PluginRoutes +from module.security.api import get_current_user + +from .deps import get_context + +router = APIRouter(prefix="/plugins", tags=["plugins"]) + +_METHODS = ["GET", "POST", "PUT", "PATCH", "DELETE", "HEAD", "OPTIONS"] + + +class _Forward(Response): + """把当前请求原样交给插件路由，由插件路由自己发送响应。 + + 请求体未被读取（端点不声明 body 参数），插件路由可以正常读取。 + """ + + def __init__(self, routes: PluginRoutes, plugin_id: str, sub_path: str) -> None: + super().__init__() + self._routes = routes + self._plugin_id = plugin_id + self._sub_path = sub_path + + async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: + path: str = scope["path"] + # 与 Starlette Mount 一致：path 保持完整，root_path 指向挂载前缀， + # 插件路由按 ``path - root_path`` 匹配（即 /api/v1/plugins/ 之后的部分） + child = { + k: v + for k, v in scope.items() + if k not in ("route", "endpoint", "path_params", "router") + } + child["root_path"] = path[: len(path) - len(self._sub_path)] + handled = await self._routes.dispatch(self._plugin_id, child, receive, send) + if not handled: + raise HTTPException(status_code=404, detail="Not Found") + + +@router.api_route( + "/{plugin_id}/{path:path}", + methods=_METHODS, + dependencies=[Depends(get_current_user)], + include_in_schema=False, +) +async def plugin_route( + plugin_id: str, path: str, ctx: AppContext = Depends(get_context) +): + """转发到插件 ``plugin_id`` 通过 ``points.API_ROUTER`` 提供的路由。""" + routes = ctx.plugin_routes + if plugin_id not in routes.plugin_ids(): + raise HTTPException( + status_code=404, detail=f"Unknown plugin route: {plugin_id}" + ) + return _Forward(routes, plugin_id, "/" + path) diff --git a/backend/src/module/core/context.py b/backend/src/module/core/context.py index a6559a46..dbc2f360 100644 --- a/backend/src/module/core/context.py +++ b/backend/src/module/core/context.py @@ -20,8 +20,7 @@ from module.parser.analyser.mikan_parser import reset_cache as reset_mikan_cache from module.parser.analyser.tmdb_parser import reset_cache as reset_tmdb_cache from module.parser.title_parser import reset_cache as reset_llm_parser from module.plugin import PluginManager -from module.plugin import host as plugin_host -from module.plugin.host import get_registry +from module.plugin.host import get_registry, set_bus, set_runner from module.rss import RSSAnalyser from module.searcher.searcher import reset_cache as reset_poster_cache from module.update import ( @@ -38,6 +37,7 @@ from .loops import ( rss_tick, update_check_tick, ) +from .plugin_routes import PluginRoutes from .plugin_tasks import PluginTasks from .scheduler import PeriodicTask, Scheduler @@ -90,10 +90,12 @@ class AppContext: self._plugin_tasks = PluginTasks( scheduler, self.plugins.registry, self.plugins.breaker ) + self.plugin_routes = PluginRoutes(self.plugins.registry, self.plugins.breaker) self.plugins.on_change = self._on_plugins_changed - # 模块级代码（RSS 引擎、下载门面、网络层）没有 ctx，经进程级访问器 - # 拿到 runner 执行 ingest 钩子 - plugin_host.set_runner(self.plugins.runner) + # 宿主各处（通知管理器、站内通知中心、流水线）经 module.plugin.host 发布 + # 事件与执行钩子，统一指向本进程的插件管理器 + set_bus(self.plugins.bus) + set_runner(self.plugins.runner) # Downloader-status TTL cache (was ProgramStatus.check_downloader_status). self._downloader_status = False self._downloader_reason: str | None = None @@ -112,6 +114,7 @@ class AppContext: async def _on_plugins_changed(self) -> None: """插件加载/重载/停用后，重建依赖插件 Provider 的派生状态。""" await self._plugin_tasks.sync() + self.plugin_routes.sync() # 插件提供的通知渠道在插件加载后才可解析；LLM 解析器可能缓存了旧的 # 插件适配器。下载器客户端缓存键包含 Provider 登记项，会自动换新。 self.notifier.rebuild() diff --git a/backend/src/module/core/plugin_routes.py b/backend/src/module/core/plugin_routes.py new file mode 100644 index 00000000..727034c5 --- /dev/null +++ b/backend/src/module/core/plugin_routes.py @@ -0,0 +1,92 @@ +"""插件 REST 路由（points.API_ROUTER）的挂载表。 + +FastAPI 不支持在运行时卸载路由，因此宿主只注册一个固定的分发路由 +``/api/v1/plugins/{plugin_id}/{path:path}``（见 ``module/api/plugin_routes.py``， +鉴权在那里强制），再按 plugin_id 转发给这里维护的、由插件 ``APIRouter`` 组成的 +ASGI 应用。插件加载/重载/停用后 :meth:`PluginRoutes.sync` 重建该表，路由随插件 +出现和消失。 +""" + +import logging + +from fastapi import APIRouter +from starlette.exceptions import HTTPException +from starlette.types import ASGIApp, Receive, Scope, Send + +from ab_sdk import points +from module.plugin.registry import ExtensionRegistry, ProviderEntry +from module.plugin.runner import CircuitBreaker + +logger = logging.getLogger(__name__) + + +class PluginRoutes: + def __init__(self, registry: ExtensionRegistry, breaker: CircuitBreaker) -> None: + self._registry = registry + self._breaker = breaker + # plugin_id -> (构建时的 Provider 登记项, 合并后的路由) + self._apps: dict[str, tuple[tuple[ProviderEntry, ...], APIRouter]] = {} + + def sync(self) -> None: + """按注册表重建：插件重载后登记项是新对象，据此替换为新实例的路由。""" + grouped: dict[str, list[ProviderEntry]] = {} + for entry in self._registry.providers(points.API_ROUTER).values(): + grouped.setdefault(entry.plugin_id, []).append(entry) + apps: dict[str, tuple[tuple[ProviderEntry, ...], APIRouter]] = {} + for plugin_id, entries in grouped.items(): + key = tuple(sorted(entries, key=lambda e: e.id)) + current = self._apps.get(plugin_id) + if current is not None and current[0] == key: + apps[plugin_id] = current + continue + router = self._build(plugin_id, key) + if router is not None: + apps[plugin_id] = (key, router) + self._apps = apps + + def plugin_ids(self) -> list[str]: + return sorted(self._apps) + + def _build( + self, plugin_id: str, entries: tuple[ProviderEntry, ...] + ) -> APIRouter | None: + combined = APIRouter() + for entry in entries: + try: + router = entry.factory() + if not isinstance(router, APIRouter): + raise TypeError(f"应返回 fastapi.APIRouter，收到 {type(router)}") + combined.include_router(router) + except Exception as e: + reason = f"{points.API_ROUTER} {entry.id}: {type(e).__name__}: {e}" + logger.warning("[Plugin:%s] 路由创建失败：%s", plugin_id, reason) + self._breaker.record_failure(plugin_id, reason) + return None + logger.debug("[Plugin:%s] 已挂载 %d 条路由", plugin_id, len(combined.routes)) + return combined + + async def dispatch( + self, plugin_id: str, scope: Scope, receive: Receive, send: Send + ) -> bool: + """把请求交给插件路由处理；插件没有路由时返回 False（由调用方回 404）。 + + ``scope`` 应已按 Starlette ``Mount`` 的约定设置好 ``root_path``，插件 + 路由据此只看到 ``/api/v1/plugins/`` 之后的路径。插件处理函数抛出的 + 异常计入熔断后继续向上抛，由宿主的异常中间件返回 500。 + """ + found = self._apps.get(plugin_id) + if found is None: + return False + app: ASGIApp = found[1] + try: + await app(scope, receive, send) + except HTTPException: + # 未匹配的路径（404/405）或插件主动抛出的 HTTP 错误，不算插件故障 + raise + except Exception as e: + self._breaker.record_failure( + plugin_id, f"{points.API_ROUTER}: {type(e).__name__}: {e}" + ) + raise + self._breaker.record_success(plugin_id) + return True diff --git a/backend/src/module/mcp/plugins.py b/backend/src/module/mcp/plugins.py new file mode 100644 index 00000000..1ca923b3 --- /dev/null +++ b/backend/src/module/mcp/plugins.py @@ -0,0 +1,182 @@ +"""插件提供的 MCP 工具与资源（points.MCP_TOOL / points.MCP_RESOURCE）。 + +每次 list / call 都从扩展注册表现取，插件加载、重载、停用后立即生效。对外 +名称加插件 id 前缀以避免与内置及其它插件冲突： + +- 工具名 ``__``：不少 MCP 客户端会把工具名原样交给 LLM API， + 而 Anthropic / OpenAI 的工具名只接受 ``^[a-zA-Z0-9_-]{1,64}$``，所以不用 ``.``。 + 插件 id 只含小写字母、数字与连字符，第一个 ``__`` 即为分隔符，不会歧义。 +- 资源 URI ``autobangumi://plugins//``。 + +插件处理函数失败或超时会计入该插件的熔断计数。 +""" + +import asyncio +import logging +import re +from collections.abc import Callable +from typing import Any + +from mcp import types +from pydantic import AnyUrl + +from ab_sdk import points +from ab_sdk.mcp import DEFAULT_TOOL_TIMEOUT, McpResource, McpTool +from module.plugin.host import get_registry +from module.plugin.registry import ProviderEntry + +from .runtime import get_context + +logger = logging.getLogger(__name__) + +TOOL_SEPARATOR = "__" +RESOURCE_PREFIX = "autobangumi://plugins/" +_TOOL_NAME = re.compile(r"^[A-Za-z0-9_-]{1,64}$") + + +def _record(plugin_id: str, reason: str | None) -> None: + """把调用结果计入插件熔断（MCP 运行时未绑定 AppContext 时跳过）。""" + ctx = get_context() + if ctx is None: + return + breaker = ctx.plugins.breaker + if reason is None: + breaker.record_success(plugin_id) + else: + breaker.record_failure(plugin_id, reason) + + +def _resolve( + point: str, expected: type, spec: Callable[[ProviderEntry, Any], Any] +) -> list[tuple[ProviderEntry, Any, Any]]: + """调用该扩展点的全部插件工厂并用 ``spec`` 生成对外的 MCP 描述， + 跳过失败、类型不符或字段不合法（MCP 模型校验失败）的条目。""" + resolved = [] + for entry in get_registry().providers(point).values(): + try: + impl = entry.factory() + if not isinstance(impl, expected): + raise TypeError(f"应返回 {expected.__name__}，收到 {type(impl)}") + described = spec(entry, impl) + except Exception as e: + reason = f"{point} {entry.id}: {type(e).__name__}: {e}" + logger.warning("[Plugin:%s] %s", entry.plugin_id, reason) + _record(entry.plugin_id, reason) + continue + resolved.append((entry, impl, described)) + return resolved + + +# ------------------------------------------------------------------ tools + + +def tool_name(plugin_id: str, tool_id: str) -> str: + return f"{plugin_id}{TOOL_SEPARATOR}{tool_id}" + + +def _tool_spec(entry: ProviderEntry, tool: McpTool) -> types.Tool: + return types.Tool( + name=tool_name(entry.plugin_id, entry.id), + description=tool.description, + inputSchema=tool.input_schema, + ) + + +def _plugin_tools() -> dict[str, tuple[ProviderEntry, McpTool, types.Tool]]: + tools = {} + for entry, tool, spec in _resolve(points.MCP_TOOL, McpTool, _tool_spec): + name = spec.name + if not _TOOL_NAME.match(name): + logger.warning( + "[Plugin:%s] MCP 工具名 %r 不合法（仅限字母、数字、_、-，" + "不超过 64 字符），已跳过", + entry.plugin_id, + name, + ) + continue + tools[name] = (entry, tool, spec) + return tools + + +def list_plugin_tools() -> list[types.Tool]: + return [spec for _, (_, _, spec) in sorted(_plugin_tools().items())] + + +async def call_plugin_tool(name: str, arguments: dict) -> tuple[bool, Any]: + """调用插件工具；``name`` 不是插件工具时返回 ``(False, None)``。 + + 插件处理函数的异常会继续抛出（由 ``handle_tool`` 统一转成错误 JSON）， + 在此之前计入熔断。 + """ + if TOOL_SEPARATOR not in name: + return False, None + found = _plugin_tools().get(name) + if found is None: + return False, None + entry, tool, _ = found + try: + result = await asyncio.wait_for(tool.handler(arguments), tool.timeout) + except Exception as e: + reason = ( + f"MCP 工具 {entry.id} 超时（{tool.timeout}s）" + if isinstance(e, TimeoutError) + else f"MCP 工具 {entry.id}: {type(e).__name__}: {e}" + ) + _record(entry.plugin_id, reason) + raise + _record(entry.plugin_id, None) + return True, result + + +# ------------------------------------------------------------------ resources + + +def resource_uri(plugin_id: str, resource_id: str) -> str: + return f"{RESOURCE_PREFIX}{plugin_id}/{resource_id}" + + +def _resource_spec(entry: ProviderEntry, resource: McpResource) -> types.Resource: + return types.Resource( + uri=AnyUrl(resource_uri(entry.plugin_id, entry.id)), + name=resource.name, + description=resource.description or None, + mimeType=resource.mime_type, + ) + + +def _plugin_resources() -> dict[str, tuple[ProviderEntry, McpResource, types.Resource]]: + # 以 AnyUrl 规范化后的 URI 为键：含空格或非 ASCII 字符的 id 会被百分号编码， + # 客户端读取时传回的正是列表里的编码形式 + return { + str(spec.uri): (entry, resource, spec) + for entry, resource, spec in _resolve( + points.MCP_RESOURCE, McpResource, _resource_spec + ) + } + + +def list_plugin_resources() -> list[types.Resource]: + return [spec for _, (_, _, spec) in sorted(_plugin_resources().items())] + + +async def read_plugin_resource(uri: str) -> tuple[bool, Any]: + """读取插件资源；``uri`` 不是插件资源时返回 ``(False, None)``。""" + if not uri.startswith(RESOURCE_PREFIX): + return False, None + found = _plugin_resources().get(uri) + if found is None: + return False, None + entry, resource, _ = found + # McpResource 没有超时字段，沿用工具的默认超时 + try: + result = await asyncio.wait_for(resource.handler(), DEFAULT_TOOL_TIMEOUT) + except Exception as e: + reason = ( + f"MCP 资源 {entry.id} 超时（{DEFAULT_TOOL_TIMEOUT}s）" + if isinstance(e, TimeoutError) + else f"MCP 资源 {entry.id}: {type(e).__name__}: {e}" + ) + _record(entry.plugin_id, reason) + raise + _record(entry.plugin_id, None) + return True, result diff --git a/backend/src/module/mcp/resources.py b/backend/src/module/mcp/resources.py index 7e35f0b7..120f4ceb 100644 --- a/backend/src/module/mcp/resources.py +++ b/backend/src/module/mcp/resources.py @@ -17,6 +17,7 @@ from module.manager import TorrentManager from module.models import Bangumi from module.rss import RSSEngine +from .plugins import list_plugin_resources, read_plugin_resource from .runtime import get_context from .tools import _bangumi_to_dict @@ -53,6 +54,11 @@ RESOURCE_TEMPLATES = [ ] +def all_resources() -> list[types.Resource]: + """内置资源 + 当前已启用插件提供的资源（``autobangumi://plugins//...``）。""" + return RESOURCES + list_plugin_resources() + + async def handle_resource(uri: str) -> str: """Return a JSON string for the given MCP resource URI. @@ -61,6 +67,7 @@ async def handle_resource(uri: str) -> str: - ``autobangumi://status`` - program version and running state - ``autobangumi://rss/feeds`` - configured RSS feeds - ``autobangumi://anime/{id}`` - single anime by integer ID + - ``autobangumi://plugins//`` - 插件提供的资源 """ if uri == "autobangumi://anime/list": async with Database() as db: @@ -108,4 +115,14 @@ async def handle_resource(uri: str) -> str: return json.dumps(_bangumi_to_dict(result), ensure_ascii=False) return json.dumps({"error": result.msg_en}) + try: + handled, content = await read_plugin_resource(uri) + except Exception as e: + logger.warning("Plugin resource %s failed: %s", uri, e) + return json.dumps({"error": str(e)}, ensure_ascii=False) + if handled: + if isinstance(content, str): + return content + return json.dumps(content, ensure_ascii=False, default=str) + return json.dumps({"error": f"Unknown resource: {uri}"}) diff --git a/backend/src/module/mcp/server.py b/backend/src/module/mcp/server.py index bf14186a..b05de20a 100644 --- a/backend/src/module/mcp/server.py +++ b/backend/src/module/mcp/server.py @@ -13,15 +13,15 @@ import logging from mcp import types from mcp.server import Server from mcp.server.sse import SseServerTransport +from pydantic import AnyUrl from starlette.applications import Starlette -from starlette.requests import Request -from starlette.responses import Response from starlette.routing import Mount, Route +from starlette.types import Receive, Scope, Send -from .resources import RESOURCE_TEMPLATES, RESOURCES, handle_resource +from .resources import RESOURCE_TEMPLATES, all_resources, handle_resource from .runtime import set_context from .security import McpAccessMiddleware -from .tools import TOOLS, handle_tool +from .tools import all_tools, handle_tool logger = logging.getLogger(__name__) @@ -31,7 +31,7 @@ sse = SseServerTransport("/messages/") @server.list_tools() async def list_tools() -> list[types.Tool]: - return TOOLS + return all_tools() @server.call_tool() @@ -42,7 +42,7 @@ async def call_tool(name: str, arguments: dict) -> list[types.TextContent]: @server.list_resources() async def list_resources() -> list[types.Resource]: - return RESOURCES + return all_resources() @server.list_resource_templates() @@ -51,22 +51,30 @@ async def list_resource_templates() -> list[types.ResourceTemplate]: @server.read_resource() -async def read_resource(uri: str) -> str: +async def read_resource(uri: AnyUrl) -> str: logger.debug("Resource read: %s", uri) - return await handle_resource(uri) + # 底层 read_resource 传入的是 pydantic AnyUrl，handle_resource 按字符串匹配 + return await handle_resource(str(uri)) -async def handle_sse(request: Request): - """Accept an SSE connection, run the MCP session until the client disconnects.""" - async with sse.connect_sse( - request.scope, request.receive, request._send - ) as streams: - await server.run( - streams[0], - streams[1], - server.create_initialization_options(), - ) - return Response() +class _SseEndpoint: + """``GET /sse`` 的裸 ASGI 端点：建立 SSE 连接并运行 MCP 会话直到客户端断开。 + + ``connect_sse`` 自己经 ``send`` 写出整个 HTTP 响应，所以这里不能是返回 + ``Response`` 的普通 endpoint 函数——Starlette 会在 SSE 结束后再发一次 + ``http.response.start``，每次断开都触发 ``AssertionError``。 + """ + + async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: + async with sse.connect_sse(scope, receive, send) as streams: + await server.run( + streams[0], + streams[1], + server.create_initialization_options(), + ) + + +handle_sse = _SseEndpoint() def create_mcp_starlette_app(ctx=None) -> Starlette: diff --git a/backend/src/module/mcp/tools.py b/backend/src/module/mcp/tools.py index 0e6d23a9..7ab891ff 100644 --- a/backend/src/module/mcp/tools.py +++ b/backend/src/module/mcp/tools.py @@ -1,5 +1,6 @@ import json import logging +from typing import Any from mcp import types @@ -11,6 +12,7 @@ from module.models import Bangumi, BangumiUpdate, RSSItem from module.rss import RSSAnalyser, RSSEngine from module.searcher import SearchTorrent +from .plugins import call_plugin_tool, list_plugin_tools from .runtime import get_context logger = logging.getLogger(__name__) @@ -175,6 +177,11 @@ TOOLS = [ ] +def all_tools() -> list[types.Tool]: + """内置工具 + 当前已启用插件提供的工具（``__``）。""" + return TOOLS + list_plugin_tools() + + def _bangumi_to_dict(b: Bangumi) -> dict: return { "id": b.id, @@ -202,7 +209,9 @@ async def handle_tool(name: str, arguments: dict) -> list[types.TextContent]: try: result = await _dispatch(name, arguments) return [ - types.TextContent(type="text", text=json.dumps(result, ensure_ascii=False)) + types.TextContent( + type="text", text=json.dumps(result, ensure_ascii=False, default=str) + ) ] except Exception as e: logger.exception("Tool %s failed", name) @@ -213,7 +222,7 @@ async def handle_tool(name: str, arguments: dict) -> list[types.TextContent]: ] -async def _dispatch(name: str, args: dict) -> dict | list: +async def _dispatch(name: str, args: dict) -> Any: if name == "list_anime": return await _list_anime(args.get("active_only", False)) elif name == "get_anime": @@ -234,8 +243,10 @@ async def _dispatch(name: str, args: dict) -> dict | list: return await _refresh_feeds() elif name == "update_anime": return await _update_anime(args) - else: - return {"error": f"Unknown tool: {name}"} + handled, result = await call_plugin_tool(name, args) + if handled: + return result + return {"error": f"Unknown tool: {name}"} async def _list_anime(active_only: bool) -> list[dict]: diff --git a/backend/src/module/notification/base.py b/backend/src/module/notification/base.py index fa2c043b..97cd1d4d 100644 --- a/backend/src/module/notification/base.py +++ b/backend/src/module/notification/base.py @@ -7,6 +7,7 @@ from module.conf import settings from module.models.bangumi import Notification from module.network import RequestContent from module.notification.events import SystemEvent +from module.notification.template import render_event if TYPE_CHECKING: from module.models.config import NotificationProvider as ProviderConfig @@ -26,6 +27,8 @@ class NotificationProvider(ABC): # 单集通知模板（{{title}}/{{season}}/{{episode}}/{{poster_url}}）； # 未设置时 ``_format_message`` 回退到默认中文文案。 self.template = config.template + # 渠道类型（如 telegram），传给 message_template 钩子按渠道定制文案 + self.channel = str(config.type).lower() async def __aenter__(self) -> "NotificationProvider": await self._http.__aenter__() @@ -62,7 +65,8 @@ class NotificationProvider(ABC): System events use each provider's default title/body delivery (:meth:`_deliver_text`) -- the per-episode template only covers {{title}}/{{season}}/{{episode}}/{{poster_url}}, which don't apply - here. + here. 标题与正文默认取 ``event.describe()``，插件可通过 + ``message_template`` 钩子按事件类型与渠道改写。 Args: event: The system event to send. @@ -70,7 +74,7 @@ class NotificationProvider(ABC): Returns: True if the event was delivered successfully, False otherwise. """ - title, body = event.describe() + title, body = await render_event(event, self.channel) return await self._deliver_text(title, body) @abstractmethod diff --git a/backend/src/module/notification/events.py b/backend/src/module/notification/events.py index 0e391f49..6830c78a 100644 --- a/backend/src/module/notification/events.py +++ b/backend/src/module/notification/events.py @@ -1,294 +1,46 @@ """非「新集数」通知的类型化事件负载。 刻意与 ``Notification``（module/models/bangumi.py，仅描述「新集数下载完成」） -分开：RSS 订阅失败、下载添加失败、偏移量待确认三类事件携带的字段与集数通知完全 +分开：RSS 订阅失败、下载添加失败、偏移量待确认等事件携带的字段与集数通知完全 无关，硬塞进 ``Notification`` 只会制造出没有意义的 season/episode 占位值。 -每个事件同时服务两个消费方： +4.0 起事件类定义在 ``ab_sdk.events``（插件订阅时需要 import 它们），这里只做 +再导出，宿主代码的 import 路径保持不变。每个事件同时服务三个消费方： -- 外部推送（Telegram/Bark 等）：``describe() -> (标题, 正文)`` 渲染中文文案； +- 外部推送（Telegram/Bark 等）：``describe() -> (标题, 正文)`` 渲染中文文案， + 插件可通过 ``message_template`` 钩子改写； - 站内通知中心：``kind``/``severity``/``once``/``dedup_key()``/``payload()`` - 提供结构化字段，前端按 kind + payload 做多语言渲染，describe() 文案作为 - 未知 kind 的兜底展示。``once=True`` 表示同 dedup_key 终生只入库一次 - （如"新版本可用"，已读后同一版本不再提醒）。 + 提供结构化字段，前端按 kind + payload 做多语言渲染（``i18n()`` 给出 key）， + describe() 文案作为未知 kind 的兜底展示。``once=True`` 表示同 dedup_key + 终生只入库一次（如"新版本可用"，已读后同一版本不再提醒）； +- 事件总线：``NotificationManager.send_event`` 把事件发布到插件事件总线。 + +``SystemEvent`` 从封闭的 Union 改为基类：插件定义的可通知事件同样可以送进 +``send_event``。 """ -from dataclasses import dataclass -from typing import ClassVar, Optional - - -@dataclass(slots=True, frozen=True) -class RssFailureEvent: - """RSS 订阅连接状态从正常变为异常（仅状态翻转时触发一次，而非每个 tick）。""" - - kind: ClassVar[str] = "rss_failure" - severity: ClassVar[str] = "error" - once: ClassVar[bool] = False - - rss_name: str - rss_url: str - error: str - - def dedup_key(self) -> Optional[str]: - return f"rss_failure:{self.rss_url}" - - def payload(self) -> dict: - return {"rss_name": self.rss_name, "rss_url": self.rss_url, "error": self.error} - - def describe(self) -> tuple[str, str]: - """返回该事件的默认 (标题, 正文)。""" - return ( - "RSS 订阅连接异常", - f"订阅：{self.rss_name}\n地址：{self.rss_url}\n错误：{self.error}", - ) - - -@dataclass(slots=True, frozen=True) -class DownloadFailureEvent: - """匹配到的种子在（内置 HTTP 重试后）仍添加下载器失败。""" - - kind: ClassVar[str] = "download_failure" - severity: ClassVar[str] = "error" - once: ClassVar[bool] = False - - official_title: str - torrent_name: str - - def dedup_key(self) -> Optional[str]: - return f"download_failure:{self.official_title}" - - def payload(self) -> dict: - return { - "official_title": self.official_title, - "torrent_name": self.torrent_name, - } - - def describe(self) -> tuple[str, str]: - return ( - "种子添加失败", - f"番剧：{self.official_title}\n种子：{self.torrent_name}\n" - "重试后仍添加失败，请检查下载器连接。", - ) - - -@dataclass(slots=True, frozen=True) -class OffsetReviewEvent: - """番剧被标记为需要人工确认季度/集数偏移。""" - - kind: ClassVar[str] = "offset_review" - severity: ClassVar[str] = "warning" - once: ClassVar[bool] = False - - official_title: str - reason: str - - def dedup_key(self) -> Optional[str]: - return f"offset_review:{self.official_title}" - - def payload(self) -> dict: - return {"official_title": self.official_title, "reason": self.reason} - - def describe(self) -> tuple[str, str]: - return ( - "集数偏移待确认", - f"番剧：{self.official_title}\n原因：{self.reason}\n请前往设置页确认偏移量。", - ) - - -@dataclass(slots=True, frozen=True) -class DownloaderUnavailableEvent: - """下载器不可用：连不上、用户名/密码错误或 IP 被封禁。 - - ``reason`` 只进 payload 不进 dedup_key——同一台下载器从 unreachable 翻转 - 到 credentials 时合并为一条（内容以最新为准），不产生两行。 - """ - - kind: ClassVar[str] = "downloader_unavailable" - severity: ClassVar[str] = "error" - once: ClassVar[bool] = False - - host: str - reason: str # unreachable | credentials | banned - - def dedup_key(self) -> Optional[str]: - return f"downloader:{self.host}" - - def payload(self) -> dict: - return {"host": self.host, "reason": self.reason} - - def describe(self) -> tuple[str, str]: - details = { - "credentials": "用户名或密码错误，请在设置中检查下载器凭据。", - "banned": "IP 已被下载器封禁，请在下载器 WebUI 中解封或重启下载器。", - "unreachable": "无法连接下载器，请检查地址、端口和网络。", - } - detail = details.get(self.reason, details["unreachable"]) - return ("下载器连接异常", f"下载器：{self.host}\n{detail}") - - -@dataclass(slots=True, frozen=True) -class UpdateAvailableEvent: - """检查到可用的新版本。""" - - kind: ClassVar[str] = "update_available" - severity: ClassVar[str] = "info" - once: ClassVar[bool] = True - - current: str - latest: str - channel: str - notes: str = "" - - def dedup_key(self) -> Optional[str]: - return f"update_available:{self.latest}" - - def payload(self) -> dict: - return { - "current": self.current, - "latest": self.latest, - "channel": self.channel, - "notes": self.notes, - } - - def describe(self) -> tuple[str, str]: - return ( - "发现新版本", - f"当前版本：{self.current}\n最新版本：{self.latest}（{self.channel} 频道）\n" - "可前往 设置 → 软件更新 升级。", - ) - - -@dataclass(slots=True, frozen=True) -class UpdateAppliedEvent: - """更新应用结果（成功或失败）。每次结果都单独入库，不做去重。""" - - once: ClassVar[bool] = False - - version: str - success: bool - message: str = "" - - @property - def kind(self) -> str: - return "update_applied" if self.success else "update_failed" - - @property - def severity(self) -> str: - return "info" if self.success else "error" - - def dedup_key(self) -> Optional[str]: - return None - - def payload(self) -> dict: - return {"version": self.version, "message": self.message} - - def describe(self) -> tuple[str, str]: - if self.success: - return ("程序更新完成", f"已更新到 {self.version}，重启后生效。") - return ("程序更新失败", f"版本：{self.version}\n原因：{self.message}") - - -@dataclass(slots=True, frozen=True) -class LLMAuthFailureEvent: - """订阅类 LLM 提供商凭据失效（刷新失败），需要用户重新连接。""" - - kind: ClassVar[str] = "llm_auth_failure" - severity: ClassVar[str] = "error" - once: ClassVar[bool] = False - - provider_id: str - account_label: str = "" - message: str = "" - - def dedup_key(self) -> Optional[str]: - return f"llm_auth:{self.provider_id}" - - def payload(self) -> dict: - return { - "provider_id": self.provider_id, - "account_label": self.account_label, - "message": self.message, - } - - def describe(self) -> tuple[str, str]: - account = f"（{self.account_label}）" if self.account_label else "" - return ( - "LLM 提供商凭据失效", - f"提供商：{self.provider_id}{account}\n{self.message}\n" - "请前往 设置 → LLM 解析器 重新连接。", - ) - - -@dataclass(slots=True, frozen=True) -class LLMPluginInstallFailedEvent: - """LLM 提供商插件安装失败（下载/签名校验/兼容性）。""" - - kind: ClassVar[str] = "llm_plugin_install_failed" - severity: ClassVar[str] = "error" - once: ClassVar[bool] = False - - plugin_id: str - version: str = "" - message: str = "" - - def dedup_key(self) -> Optional[str]: - return f"llm_plugin_install:{self.plugin_id}" - - def payload(self) -> dict: - return { - "plugin_id": self.plugin_id, - "version": self.version, - "message": self.message, - } - - def describe(self) -> tuple[str, str]: - return ( - "LLM 插件安装失败", - f"插件：{self.plugin_id} {self.version}\n原因：{self.message}", - ) - - -@dataclass(slots=True, frozen=True) -class RenameConflictEvent: - """A media rename reached a durable target-path conflict.""" - - kind: ClassVar[str] = "rename_conflict" - severity: ClassVar[str] = "warning" - once: ClassVar[bool] = False - - task_id: str - torrent_name: str - target_path: str - reason: str - - def dedup_key(self) -> Optional[str]: - return f"rename_conflict:{self.task_id}:{self.target_path}" - - def payload(self) -> dict: - return { - "task_id": self.task_id, - "torrent_name": self.torrent_name, - "target_path": self.target_path, - "reason": self.reason, - } - - def describe(self) -> tuple[str, str]: - return ( - "媒体文件重命名冲突", - f"种子：{self.torrent_name}\n目标：{self.target_path}\n" - f"原因：{self.reason}", - ) - - -# 除「新集数」以外的通知事件的联合类型。 -SystemEvent = ( - RssFailureEvent - | DownloadFailureEvent - | OffsetReviewEvent - | DownloaderUnavailableEvent - | UpdateAvailableEvent - | UpdateAppliedEvent - | LLMAuthFailureEvent - | LLMPluginInstallFailedEvent - | RenameConflictEvent +from ab_sdk.events import ( + DownloaderUnavailableEvent, + DownloadFailureEvent, + LLMAuthFailureEvent, + LLMPluginInstallFailedEvent, + OffsetReviewEvent, + RenameConflictEvent, + RssFailureEvent, + SystemEvent, + UpdateAppliedEvent, + UpdateAvailableEvent, ) + +__all__ = [ + "DownloadFailureEvent", + "DownloaderUnavailableEvent", + "LLMAuthFailureEvent", + "LLMPluginInstallFailedEvent", + "OffsetReviewEvent", + "RenameConflictEvent", + "RssFailureEvent", + "SystemEvent", + "UpdateAppliedEvent", + "UpdateAvailableEvent", +] diff --git a/backend/src/module/notification/inbox.py b/backend/src/module/notification/inbox.py index 31137765..71b9823c 100644 --- a/backend/src/module/notification/inbox.py +++ b/backend/src/module/notification/inbox.py @@ -2,15 +2,20 @@ ``record_event`` 把 SystemEvent 落库（``async with Database()``， session-per-operation）；``inbox_revision`` 是进程内单调递增的修订号， -任何入库/已读/删除操作都会 bump，SSE 循环据此决定是否需要查库推送， -避免每个 tick 都打一次数据库。AB 为单进程部署，进程内计数即可。 +任何入库/已读/删除操作都会 bump，并在事件总线上发布 :class:`InboxChanged`。 +SSE 连接订阅该事件，收到后才查库推送未读数（取代原先每 3s 比较修订号的轮询）。 +AB 为单进程部署，进程内计数即可。 """ import json import logging +from dataclasses import dataclass +from typing import ClassVar +from ab_sdk import Event from module.database import Database from module.notification.events import SystemEvent +from module.plugin.host import publish logger = logging.getLogger(__name__) @@ -19,6 +24,14 @@ INBOX_KEEP = 500 _revision: int = 1 +@dataclass(frozen=True, slots=True) +class InboxChanged(Event): + """站内通知中心有变化（新消息、已读、删除），``revision`` 为新的修订号。""" + + kind: ClassVar[str] = "inbox.changed" + revision: int + + def inbox_revision() -> int: return _revision @@ -26,6 +39,7 @@ def inbox_revision() -> int: def bump_inbox_revision() -> None: global _revision _revision += 1 + publish(InboxChanged(revision=_revision)) async def record_event(event: SystemEvent) -> int: diff --git a/backend/src/module/notification/manager.py b/backend/src/module/notification/manager.py index b962ff6e..862d7450 100644 --- a/backend/src/module/notification/manager.py +++ b/backend/src/module/notification/manager.py @@ -9,6 +9,7 @@ from module.database import Database from module.models.bangumi import Notification from module.notification.events import SystemEvent from module.notification.inbox import record_event +from module.plugin.host import publish if TYPE_CHECKING: from module.models.config import NotificationProvider as ProviderConfig @@ -118,8 +119,9 @@ class NotificationManager: async def send_event(self, event: SystemEvent): """Persist a system event to the in-app inbox, then broadcast it. - 持久化不受 ``settings.notification.enable`` 影响（该开关只管外部 - 推送），失败也不阻塞外部广播。 + 顺序：写入站内通知中心 → 发布到插件事件总线 → 推送到外部渠道。 + 持久化与总线发布不受 ``settings.notification.enable`` 影响（该开关只管 + 外部推送），持久化失败也不阻塞后两步。 Args: event: The system event to send. @@ -132,6 +134,8 @@ class NotificationManager: type(event).__name__, exc_info=True, ) + # 插件 @subscribe(event.kind) 的订阅者在各自队列中异步处理，不阻塞这里 + publish(event) if not settings.notification.enable: return diff --git a/backend/src/module/notification/template.py b/backend/src/module/notification/template.py new file mode 100644 index 00000000..5fd2aec0 --- /dev/null +++ b/backend/src/module/notification/template.py @@ -0,0 +1,37 @@ +"""系统事件推送文案：默认 ``describe()``，经 ``message_template`` 钩子改写。""" + +import logging + +from ab_sdk import points +from ab_sdk.notify import RenderedMessage +from module.notification.events import SystemEvent +from module.plugin.host import get_runner + +logger = logging.getLogger(__name__) + + +async def render_event(event: SystemEvent, channel: str) -> tuple[str, str]: + """返回推送到 ``channel`` 渠道的 (标题, 正文)。 + + 没有插件挂 ``message_template`` 钩子时与 ``event.describe()`` 完全一致； + 钩子失败、超时或返回值不是 ``RenderedMessage`` 时沿用上一步的结果（并计入 + 该插件的熔断）。 + """ + title, body = event.describe() + runner = get_runner() + if runner is None: + return title, body + try: + message = await runner.transform( + points.MESSAGE_TEMPLATE, + RenderedMessage(title, body), + event, + channel, + expect=RenderedMessage, + ) + except Exception: + # 单个钩子的失败已在 runner 内隔离；这里兜底的是运行时本身的异常 + # （如扩展点未声明），任何情况下都不能让通知因模板而发不出去 + logger.warning("[Notification] 渲染通知模板失败，使用默认文案", exc_info=True) + return title, body + return message.title, message.body diff --git a/backend/src/module/plugin/host.py b/backend/src/module/plugin/host.py index 62652158..8f3d4159 100644 --- a/backend/src/module/plugin/host.py +++ b/backend/src/module/plugin/host.py @@ -12,9 +12,10 @@ import logging from collections.abc import Callable from typing import Any -from ab_sdk import points +from ab_sdk import Event, points from ab_sdk.downloader import DownloaderConnection +from .bus import EventBus from .registry import ExtensionPoint, ExtensionRegistry, ProviderEntry from .runner import HookRunner @@ -40,10 +41,23 @@ POINTS = ( ExtensionPoint(points.TITLE_PARSED, "transform", "修正标题解析结果"), ExtensionPoint(points.TORRENT_ADDING, "transform", "修改发给下载器的添加请求"), ExtensionPoint(points.HTTP_REQUEST, "transform", "修改宿主 GET 请求的请求头"), + # --- P5 events/api --- + ExtensionPoint( + points.API_ROUTER, + "provider", + "插件 REST 路由（/api/v1/plugins//）", + scoped=True, + ), + ExtensionPoint(points.MCP_TOOL, "provider", "MCP 工具", scoped=True), + ExtensionPoint(points.MCP_RESOURCE, "provider", "MCP 资源", scoped=True), + ExtensionPoint(points.MESSAGE_TEMPLATE, "transform", "系统事件通知文案"), ) _registry: ExtensionRegistry | None = None -# 由 AppContext 在构建时设置；未设置时（测试、CLI）钩子一律跳过，只执行宿主逻辑 +# 进程级事件总线与钩子执行器，由 AppContext 在构造时设置（即 PluginManager +# 持有的那一份）；未设置时（单元测试、脚本、CLI）发布事件为空操作、钩子一律 +# 跳过，只执行宿主逻辑 +_bus: EventBus | None = None _runner: HookRunner | None = None @@ -159,3 +173,30 @@ def _register_core(registry: ExtensionRegistry) -> None: points.METADATA_PROVIDER, ProviderEntry(CORE, provider_id, metadata(provider_id)), ) + + +# --- P5 events/api --------------------------------------------------------- + + +def get_bus() -> EventBus | None: + return _bus + + +def set_bus(bus: EventBus | None) -> None: + global _bus + _bus = bus + + +def publish(event: Event) -> None: + """把宿主事件发布到进程级总线；总线未设置时什么也不做。 + + 发布只负责入队，订阅者的失败不会影响调用方。没有运行中的事件循环时 + （同步上下文）无法投递，记录日志后忽略。 + """ + bus = _bus + if bus is None: + return + try: + bus.publish(event) + except RuntimeError: + logger.debug("[EventBus] 无事件循环，丢弃事件 %s", event.kind) diff --git a/backend/src/module/plugin/registry.py b/backend/src/module/plugin/registry.py index 374de132..01d23af4 100644 --- a/backend/src/module/plugin/registry.py +++ b/backend/src/module/plugin/registry.py @@ -18,6 +18,9 @@ class ExtensionPoint: description: str = "" # 仅 filter：钩子出错/超时时是否放行（True）还是视为拒绝（False） fail_open: bool = True + # 仅 provider：True 时 Provider id 只需在插件内唯一（宿主对外暴露时会加 + # 插件 id 前缀，如插件路由、MCP 工具），注册表按 ``/`` 存储 + scoped: bool = False @dataclass(frozen=True, slots=True) @@ -68,13 +71,14 @@ class ExtensionRegistry: if point.kind != "provider": raise RegistryError(f"{point_name} 是 {point.kind} 扩展点，请使用 @hook") slot = self._providers.setdefault(point_name, {}) - owner = slot.get(entry.id) + key = f"{entry.plugin_id}/{entry.id}" if point.scoped else entry.id + owner = slot.get(key) if owner is not None: raise RegistryError( f"{point_name} 的 Provider id {entry.id!r} 已被插件 " f"{owner.plugin_id} 占用" ) - slot[entry.id] = entry + slot[key] = entry def remove_plugin(self, plugin_id: str) -> None: for name, entries in self._hooks.items(): @@ -100,5 +104,6 @@ class ExtensionRegistry: return bool(self._hooks.get(point_name)) def providers(self, point_name: str) -> dict[str, ProviderEntry]: + """按 Provider id 索引；``scoped`` 扩展点的键为 ``/``。""" self.point(point_name) return dict(self._providers.get(point_name, {})) diff --git a/backend/src/test/test_api_events.py b/backend/src/test/test_api_events.py index 9a14c7e7..36acdbf7 100644 --- a/backend/src/test/test_api_events.py +++ b/backend/src/test/test_api_events.py @@ -214,7 +214,18 @@ class TestLogPayload: class TestNotificationEvent: - def _generator(self, tmp_path): + """notification 帧由事件总线上的 inbox.changed 驱动（P5），线格式不变。""" + + @pytest.fixture + def bus(self, monkeypatch): + from module.plugin import host + from module.plugin.bus import EventBus + + bus = EventBus() + monkeypatch.setattr(host, "_bus", bus) + return bus + + def _generator(self, tmp_path, *, tick=0.01, payload=None): from module.api.events import _event_generator request = AsyncMock() @@ -222,17 +233,17 @@ class TestNotificationEvent: ctx = MagicMock() ctx.is_running = True ctx.first_run = False + notification = payload or AsyncMock( + return_value={"unread_count": 2, "latest_id": 9} + ) patches = ( patch( "module.api.events._downloader_payload", new=AsyncMock(return_value=None), ), patch("module.api.events.LOG_PATH", tmp_path / "missing.log"), - patch( - "module.api.events._notification_payload", - new=AsyncMock(return_value={"unread_count": 2, "latest_id": 9}), - ), - patch("module.api.events.asyncio.sleep", new=AsyncMock()), + patch("module.api.events._notification_payload", new=notification), + patch("module.api.events._TICK_SECONDS", tick), ) return _event_generator(request, ctx), patches @@ -242,7 +253,7 @@ class TestNotificationEvent: events.append(await asyncio.wait_for(gen.__anext__(), timeout=1.0)) return events - async def test_first_tick_emits_notification_frame(self, tmp_path): + async def test_first_tick_emits_notification_frame(self, tmp_path, bus): from module.notification.inbox import inbox_revision gen, patches = self._generator(tmp_path) @@ -252,14 +263,12 @@ class TestNotificationEvent: finally: await gen.aclose() - by_name = {e["event"]: e for e in events} - assert set(by_name) == {"status", "downloader", "notification"} - data = json.loads(by_name["notification"]["data"]) - assert data["unread_count"] == 2 - assert data["latest_id"] == 9 - assert data["revision"] == inbox_revision() + assert [e["event"] for e in events] == ["status", "downloader", "notification"] + # 线格式与 3.x 一致：unread_count / latest_id / revision 三个字段 + data = json.loads(events[2]["data"]) + assert data == {"unread_count": 2, "latest_id": 9, "revision": inbox_revision()} - async def test_no_frame_when_revision_unchanged(self, tmp_path): + async def test_no_frame_without_inbox_change(self, tmp_path, bus): gen, patches = self._generator(tmp_path) with patches[0], patches[1], patches[2], patches[3]: try: @@ -271,18 +280,55 @@ class TestNotificationEvent: assert [e["event"] for e in more] == ["status", "downloader"] - async def test_frame_emitted_after_revision_bump(self, tmp_path): + async def test_inbox_change_pushes_frame_immediately(self, tmp_path, bus): from module.notification.inbox import bump_inbox_revision, inbox_revision - gen, patches = self._generator(tmp_path) + # tick 设为 30s：若不是事件推送，1s 内不可能收到下一帧 + gen, patches = self._generator(tmp_path, tick=30) with patches[0], patches[1], patches[2], patches[3]: try: await self._collect(gen, 3) # 第一个 tick bump_inbox_revision() - # tick3：status + notification（修订号变了） - more = await self._collect(gen, 2) + [frame] = await self._collect(gen, 1) finally: await gen.aclose() - assert [e["event"] for e in more] == ["status", "notification"] - assert json.loads(more[1]["data"])["revision"] == inbox_revision() + assert frame["event"] == "notification" + assert json.loads(frame["data"])["revision"] == inbox_revision() + + async def test_unsubscribes_when_stream_closes(self, tmp_path, bus): + gen, patches = self._generator(tmp_path) + with patches[0], patches[1], patches[2], patches[3]: + await self._collect(gen, 3) + assert len(bus._subscribers) == 1 + await gen.aclose() + assert bus._subscribers == [] + + async def test_failed_query_is_retried_next_tick(self, tmp_path, bus): + payload = AsyncMock( + side_effect=[RuntimeError("db busy"), {"unread_count": 1, "latest_id": 3}] + ) + gen, patches = self._generator(tmp_path, payload=payload) + with patches[0], patches[1], patches[2], patches[3]: + try: + events = await self._collect(gen, 3) + finally: + await gen.aclose() + + # tick0：status + downloader（notification 查库失败）；tick1 重试成功 + assert [e["event"] for e in events] == ["status", "downloader", "notification"] + assert json.loads(events[2]["data"])["latest_id"] == 3 + assert payload.await_count == 2 + + async def test_without_bus_only_initial_frame(self, tmp_path, monkeypatch): + from module.plugin import host + + monkeypatch.setattr(host, "_bus", None) + gen, patches = self._generator(tmp_path) + with patches[0], patches[1], patches[2], patches[3]: + try: + events = await self._collect(gen, 5) + finally: + await gen.aclose() + + assert [e["event"] for e in events].count("notification") == 1 diff --git a/backend/src/test/test_plugin_events_api.py b/backend/src/test/test_plugin_events_api.py new file mode 100644 index 00000000..8fa6d287 --- /dev/null +++ b/backend/src/test/test_plugin_events_api.py @@ -0,0 +1,695 @@ +"""P5：系统事件上总线、message_template、插件 REST 路由与 MCP 工具/资源。""" + +import json +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from fastapi import APIRouter, FastAPI, HTTPException +from fastapi.testclient import TestClient +from pydantic import BaseModel + +from ab_sdk import Event, Plugin, hook, points, provider, subscribe +from ab_sdk.events import ( + RssFailureEvent, + SystemEvent, + UpdateAppliedEvent, +) +from ab_sdk.mcp import McpResource, McpTool +from ab_sdk.notify import RenderedMessage +from module.api import v1 +from module.api.deps import get_context +from module.core.plugin_routes import PluginRoutes +from module.models.config import NotificationProvider as ProviderConfig +from module.models.config import Plugins +from module.notification import NotificationManager +from module.notification.inbox import InboxChanged, bump_inbox_revision +from module.notification.template import render_event +from module.plugin import host +from module.plugin.bus import EventBus +from module.plugin.loader import PluginCandidate +from module.plugin.manager import PluginManager +from module.plugin.manifest import parse_manifest +from module.plugin.registry import ExtensionRegistry, HookEntry, ProviderEntry +from module.plugin.runner import CircuitBreaker, HookRunner +from module.security.api import get_current_user + + +@pytest.fixture +def registry(monkeypatch): + monkeypatch.setattr(host, "_registry", None) + return host.get_registry() + + +@pytest.fixture +def bus(monkeypatch): + bus = EventBus() + monkeypatch.setattr(host, "_bus", bus) + return bus + + +@pytest.fixture +def breaker(): + return CircuitBreaker() + + +@pytest.fixture +def runner(monkeypatch, registry, breaker): + runner = HookRunner(registry, breaker) + monkeypatch.setattr(host, "_runner", runner) + return runner + + +def add(registry, point, provider_id, impl, plugin_id="ext"): + registry.add_provider(point, ProviderEntry(plugin_id, provider_id, lambda: impl)) + + +# ---------------------------------------------------------------- system events + + +class TestSystemEvents: + def test_system_events_are_sdk_events(self): + event = RssFailureEvent(rss_name="Feed", rss_url="http://x", error="boom") + assert isinstance(event, Event) + assert isinstance(event, SystemEvent) + assert event.kind == "rss_failure" + # 宿主旧的 import 路径仍然可用，且是同一个类 + from module.notification import RssFailureEvent as HostRssFailureEvent + + assert HostRssFailureEvent is RssFailureEvent + + def test_i18n_key_matches_frontend_and_payload_is_params(self): + event = RssFailureEvent(rss_name="Feed", rss_url="http://x", error="boom") + key, params = event.i18n() + assert key == "notifications.kind.rss_failure" + assert params == event.payload() + failed = UpdateAppliedEvent(version="4.0.0", success=False, message="x") + assert failed.i18n()[0] == "notifications.kind.update_failed" + + def test_plugin_defined_system_event_defaults(self): + from dataclasses import dataclass + from typing import ClassVar + + @dataclass(frozen=True, slots=True) + class Synced(SystemEvent): + kind: ClassVar[str] = "demo.synced" + count: int + + event = Synced(count=3) + assert event.payload() == {"count": 3} + assert event.dedup_key() is None + assert event.severity == "info" + assert event.describe() == ("demo.synced", "") + + async def test_send_event_publishes_to_bus_even_when_push_disabled(self, bus): + received: list[Event] = [] + bus.subscribe("rss_failure", received.append) + manager = NotificationManager(load_providers=False) + event = RssFailureEvent(rss_name="Feed", rss_url="http://x", error="boom") + with ( + patch( + "module.notification.manager.record_event", new_callable=AsyncMock + ) as record, + patch("module.notification.manager.settings") as mock_settings, + ): + mock_settings.notification.enable = False + await manager.send_event(event) + await bus.drain() + record.assert_awaited_once_with(event) + assert received == [event] + + async def test_send_event_publishes_even_if_inbox_write_fails(self, bus): + received: list[Event] = [] + bus.subscribe("*", received.append) + manager = NotificationManager(load_providers=False) + event = RssFailureEvent(rss_name="Feed", rss_url="http://x", error="boom") + with ( + patch( + "module.notification.manager.record_event", + new=AsyncMock(side_effect=RuntimeError("db")), + ), + patch("module.notification.manager.settings") as mock_settings, + ): + mock_settings.notification.enable = False + await manager.send_event(event) + await bus.drain() + assert received == [event] + + async def test_bump_publishes_inbox_changed(self, bus): + received: list[Event] = [] + bus.subscribe(InboxChanged.kind, received.append) + bump_inbox_revision() + await bus.drain() + [event] = received + assert isinstance(event, InboxChanged) + + def test_publish_without_bus_is_noop(self, monkeypatch): + monkeypatch.setattr(host, "_bus", None) + host.publish(InboxChanged(revision=1)) # 不抛错 + + +# ---------------------------------------------------------------- message_template + + +class TestMessageTemplate: + EVENT = RssFailureEvent(rss_name="Feed", rss_url="http://x", error="boom") + + async def test_default_is_describe_without_runner(self, monkeypatch): + monkeypatch.setattr(host, "_runner", None) + assert await render_event(self.EVENT, "telegram") == self.EVENT.describe() + + async def test_default_is_describe_without_hooks(self, runner): + assert await render_event(self.EVENT, "telegram") == self.EVENT.describe() + + async def test_hook_rewrites_per_kind_and_channel(self, runner, registry): + def render(message, event, channel): + if event.kind != "rss_failure": + return None + return RenderedMessage(f"[{channel}] {message.title}", event.rss_name) + + registry.add_hook(points.MESSAGE_TEMPLATE, HookEntry("ext", render, 100, None)) + title, body = await render_event(self.EVENT, "bark") + assert title == f"[bark] {self.EVENT.describe()[0]}" + assert body == "Feed" + other = UpdateAppliedEvent(version="1", success=True) + assert await render_event(other, "bark") == other.describe() + + async def test_bad_return_falls_back(self, runner, registry): + registry.add_hook( + points.MESSAGE_TEMPLATE, + HookEntry("ext", lambda m, e, c: "not a message", 100, None), + ) + assert await render_event(self.EVENT, "bark") == self.EVENT.describe() + + async def test_failing_hook_falls_back_and_counts(self, runner, registry, breaker): + def boom(message, event, channel): + raise RuntimeError("template bug") + + registry.add_hook(points.MESSAGE_TEMPLATE, HookEntry("ext", boom, 100, None)) + assert await render_event(self.EVENT, "bark") == self.EVENT.describe() + assert breaker._failures["ext"] == 1 + + async def test_render_event_bad_return_counts_against_its_own_plugin( + self, runner, registry, breaker + ): + def rewrite(message, event, channel): + return RenderedMessage(message.title, "rewritten") + + registry.add_hook( + points.MESSAGE_TEMPLATE, + HookEntry("bad", lambda m, e, c: "not a message", 10, None), + ) + registry.add_hook( + points.MESSAGE_TEMPLATE, HookEntry("good", rewrite, 100, None) + ) + title, body = await render_event(self.EVENT, "bark") + assert (title, body) == (self.EVENT.describe()[0], "rewritten") + assert breaker._failures.get("bad") == 1 + assert not breaker._failures.get("good") + + async def test_runtime_error_never_blocks_delivery(self, monkeypatch): + bad = MagicMock() + bad.transform = AsyncMock(side_effect=RuntimeError("undeclared point")) + monkeypatch.setattr(host, "_runner", bad) + assert await render_event(self.EVENT, "bark") == self.EVENT.describe() + + async def test_provider_send_event_uses_template(self, runner, registry): + from module.notification.providers.bark import BarkProvider + + registry.add_hook( + points.MESSAGE_TEMPLATE, + HookEntry("ext", lambda m, e, c: RenderedMessage("T", c), 100, None), + ) + provider_obj = BarkProvider( + ProviderConfig(type="bark", enabled=True, device_key="k") + ) + with patch.object( + provider_obj, "_deliver_text", new=AsyncMock(return_value=True) + ) as deliver: + assert await provider_obj.send_event(self.EVENT) + deliver.assert_awaited_once_with("T", "bark") + + +# ---------------------------------------------------------------- registry + + +def test_scoped_points_allow_same_id_in_different_plugins(registry): + add(registry, points.MCP_TOOL, "search", object(), plugin_id="a") + add(registry, points.MCP_TOOL, "search", object(), plugin_id="b") + assert set(registry.providers(points.MCP_TOOL)) == {"a/search", "b/search"} + with pytest.raises(Exception, match="已被插件"): + add(registry, points.MCP_TOOL, "search", object(), plugin_id="a") + + +# ---------------------------------------------------------------- api_router + + +def demo_router() -> APIRouter: + router = APIRouter() + + @router.get("/") + async def root(): + return {"root": True} + + @router.get("/hello") + async def hello(name: str = "world"): + return {"hello": name} + + class Body(BaseModel): + value: int + + @router.post("/echo") + async def echo(body: Body): + return {"value": body.value * 2} + + @router.get("/teapot") + async def teapot(): + raise HTTPException(status_code=418, detail="teapot") + + @router.get("/boom") + async def boom(): + raise RuntimeError("plugin bug") + + return router + + +@pytest.fixture +def routes_app(registry, breaker): + routes = PluginRoutes(registry, breaker) + plugins_mgr = MagicMock() + plugins_mgr.statuses.return_value = [] + ctx = SimpleNamespace(plugin_routes=routes, plugins=plugins_mgr) + app = FastAPI() + app.include_router(v1, prefix="/api") + app.dependency_overrides[get_context] = lambda: ctx + return app, routes + + +@pytest.fixture +def routes_client(routes_app): + app, routes = routes_app + + async def user(): + return "testuser" + + app.dependency_overrides[get_current_user] = user + return TestClient(app, raise_server_exceptions=False), routes + + +class TestApiRouter: + def test_routes_follow_plugin_lifecycle(self, routes_client, registry): + client, routes = routes_client + assert client.get("/api/v1/plugins/demo/hello").status_code == 404 + + add(registry, points.API_ROUTER, "api", demo_router(), plugin_id="demo") + routes.sync() + resp = client.get("/api/v1/plugins/demo/hello", params={"name": "ab"}) + assert resp.status_code == 200 + assert resp.json() == {"hello": "ab"} + assert client.get("/api/v1/plugins/demo/").json() == {"root": True} + + registry.remove_plugin("demo") + routes.sync() + assert client.get("/api/v1/plugins/demo/hello").status_code == 404 + + def test_body_and_errors_forwarded(self, routes_client, registry, breaker): + client, routes = routes_client + add(registry, points.API_ROUTER, "api", demo_router(), plugin_id="demo") + routes.sync() + assert client.post("/api/v1/plugins/demo/echo", json={"value": 21}).json() == { + "value": 42 + } + assert ( + client.post("/api/v1/plugins/demo/echo", json={"value": "x"}).status_code + == 422 + ) + assert client.get("/api/v1/plugins/demo/teapot").status_code == 418 + assert client.get("/api/v1/plugins/demo/missing").status_code == 404 + assert breaker._failures == {} + assert client.get("/api/v1/plugins/demo/boom").status_code == 500 + assert breaker._failures["demo"] == 1 + + def test_multiple_routers_are_merged(self, routes_client, registry): + client, routes = routes_client + extra = APIRouter() + + @extra.get("/extra") + async def extra_route(): + return {"extra": True} + + add(registry, points.API_ROUTER, "api", demo_router(), plugin_id="demo") + add(registry, points.API_ROUTER, "more", extra, plugin_id="demo") + routes.sync() + assert client.get("/api/v1/plugins/demo/extra").json() == {"extra": True} + assert client.get("/api/v1/plugins/demo/hello").status_code == 200 + + def test_bad_factory_is_skipped(self, routes_client, registry, breaker): + client, routes = routes_client + add(registry, points.API_ROUTER, "api", "not a router", plugin_id="demo") + routes.sync() + assert routes.plugin_ids() == [] + assert breaker._failures["demo"] == 1 + + def test_management_routes_not_shadowed(self, routes_client, registry): + client, routes = routes_client + add(registry, points.API_ROUTER, "api", demo_router(), plugin_id="settings") + add(registry, points.API_ROUTER, "api2", demo_router(), plugin_id="providers") + routes.sync() + assert client.get("/api/v1/plugins").status_code == 200 + providers = client.get("/api/v1/plugins/providers") + assert providers.status_code == 200 + assert "downloader" in providers.json() + # PUT /plugins/{id} 仍走插件管理路由（未知插件 → 404 且 detail 来自管理路由） + resp = client.put("/api/v1/plugins/demo", json={}) + assert resp.status_code == 404 + assert resp.json()["detail"] == "Unknown plugin: demo" + # 插件 id 恰好是 settings/providers 时，其子路径仍可访问 + assert client.get("/api/v1/plugins/settings/hello").status_code == 200 + + @patch("module.security.api.DEV_AUTH_BYPASS", False) + def test_auth_is_enforced(self, routes_app, registry): + app, routes = routes_app + add(registry, points.API_ROUTER, "api", demo_router(), plugin_id="demo") + routes.sync() + client = TestClient(app) + assert client.get("/api/v1/plugins/demo/hello").status_code == 401 + assert client.get("/api/v1/plugins/unknown/x").status_code == 401 + + +# ---------------------------------------------------------------- MCP + + +async def _search(args): + return {"q": args.get("q"), "hits": [1, 2]} + + +class TestMcp: + def test_plugin_tools_listed_with_namespace(self, registry): + from module.mcp.tools import TOOLS, all_tools + + add( + registry, + points.MCP_TOOL, + "search", + McpTool("Search", _search, {"type": "object"}), + plugin_id="demo", + ) + names = [t.name for t in all_tools()] + assert names[: len(TOOLS)] == [t.name for t in TOOLS] + assert "demo__search" in names + + async def test_call_dispatches_to_plugin(self, registry): + from module.mcp.tools import handle_tool + + add(registry, points.MCP_TOOL, "search", McpTool("S", _search), "demo") + [content] = await handle_tool("demo__search", {"q": "x"}) + assert json.loads(content.text) == {"q": "x", "hits": [1, 2]} + + async def test_unknown_names_keep_error(self, registry): + from module.mcp.tools import handle_tool + + add(registry, points.MCP_TOOL, "search", McpTool("S", _search), "demo") + for name in ("nope", "demo__nope", "other__search"): + [content] = await handle_tool(name, {}) + assert json.loads(content.text) == {"error": f"Unknown tool: {name}"} + + async def test_failure_returns_error_and_counts(self, registry, breaker): + from module.mcp.tools import handle_tool + + async def boom(args): + raise RuntimeError("plugin bug") + + add(registry, points.MCP_TOOL, "boom", McpTool("B", boom), "demo") + ctx = SimpleNamespace(plugins=SimpleNamespace(breaker=breaker)) + with patch("module.mcp.plugins.get_context", return_value=ctx): + [content] = await handle_tool("demo__boom", {}) + assert json.loads(content.text) == {"error": "plugin bug"} + assert breaker._failures["demo"] == 1 + + def test_invalid_names_and_bad_factories_skipped(self, registry): + from module.mcp.plugins import list_plugin_tools + + add(registry, points.MCP_TOOL, "has.dot", McpTool("x", _search), "demo") + add(registry, points.MCP_TOOL, "wrong", "not a tool", "demo") + add(registry, points.MCP_TOOL, "ok", McpTool("x", _search), "demo") + assert [t.name for t in list_plugin_tools()] == ["demo__ok"] + + async def test_plugin_resources(self, registry): + from module.mcp.resources import RESOURCES, all_resources, handle_resource + + async def stats(): + return {"count": 3} + + async def text(): + return "plain" + + add(registry, points.MCP_RESOURCE, "stats", McpResource("Stats", stats), "demo") + add(registry, points.MCP_RESOURCE, "text", McpResource("Text", text), "demo") + uris = [str(r.uri) for r in all_resources()] + assert uris[: len(RESOURCES)] == [str(r.uri) for r in RESOURCES] + assert "autobangumi://plugins/demo/stats" in uris + stats_json = await handle_resource("autobangumi://plugins/demo/stats") + assert json.loads(stats_json) == {"count": 3} + assert await handle_resource("autobangumi://plugins/demo/text") == "plain" + missing = await handle_resource("autobangumi://plugins/demo/none") + assert json.loads(missing) == { + "error": "Unknown resource: autobangumi://plugins/demo/none" + } + + @pytest.mark.parametrize("resource_id", ["stats", "Weekly Stats", "统计"]) + async def test_read_resource_via_protocol_returns_plugin_content( + self, registry, resource_id + ): + from mcp import types + + from module.mcp import server + + async def stats(): + return {"count": 3} + + add( + registry, + points.MCP_RESOURCE, + resource_id, + McpResource("Stats", stats), + "demo", + ) + [listed] = [ + r + for r in await server.list_resources() + if str(r.uri).startswith("autobangumi://plugins/") + ] + handler = server.server.request_handlers[types.ReadResourceRequest] + result = await handler( + types.ReadResourceRequest( + method="resources/read", + params=types.ReadResourceRequestParams(uri=listed.uri), + ) + ) + assert isinstance(result.root, types.ReadResourceResult) + [content] = result.root.contents + assert isinstance(content, types.TextResourceContents) + assert json.loads(content.text) == {"count": 3} + + @pytest.mark.parametrize( + "point, impl", + [ + (points.MCP_TOOL, McpTool("bad", _search, None)), # type: ignore[arg-type] + (points.MCP_RESOURCE, McpResource(None, _search)), # type: ignore[arg-type] + ], + ) + def test_list_malformed_plugin_spec_skipped_and_counted( + self, registry, breaker, point, impl + ): + from module.mcp.resources import RESOURCES, all_resources + from module.mcp.tools import TOOLS, all_tools + + add(registry, point, "bad", impl, "demo") + ctx = SimpleNamespace(plugins=SimpleNamespace(breaker=breaker)) + with patch("module.mcp.plugins.get_context", return_value=ctx): + assert len(all_tools()) == len(TOOLS) + assert len(all_resources()) == len(RESOURCES) + assert breaker._failures["demo"] == 1 + + async def test_read_plugin_resource_hung_handler_times_out_and_counts( + self, registry, breaker, monkeypatch + ): + import asyncio + + from module.mcp import plugins + from module.mcp.resources import handle_resource + + async def hang(): + await asyncio.Event().wait() + + monkeypatch.setattr(plugins, "DEFAULT_TOOL_TIMEOUT", 0.01) + add(registry, points.MCP_RESOURCE, "hang", McpResource("H", hang), "demo") + ctx = SimpleNamespace(plugins=SimpleNamespace(breaker=breaker)) + with patch("module.mcp.plugins.get_context", return_value=ctx): + raw = await handle_resource("autobangumi://plugins/demo/hang") + assert "error" in json.loads(raw) + assert breaker._failures["demo"] == 1 + + async def test_mcp_server_handlers_include_plugins(self, registry): + from module.mcp import server + + add(registry, points.MCP_TOOL, "search", McpTool("S", _search), "demo") + assert "demo__search" in [t.name for t in await server.list_tools()] + + +# ---------------------------------------------------------------- end to end + + +class Options(BaseModel): + greeting: str = "hi" + + +class FullPlugin(Plugin[Options]): + """一个同时使用全部 P5 扩展点的插件。""" + + config_model = Options + seen: list[Event] = [] + + @provider(points.API_ROUTER, id="api") + def api(self): + router = APIRouter() + + @router.get("/greet") + async def greet(): + return {"greeting": self.config.greeting} + + return router + + @provider(points.MCP_TOOL, id="greet") + def greet_tool(self): + async def handler(args): + return {"greeting": self.config.greeting} + + return McpTool("Greet", handler) + + @subscribe("rss_failure") + async def on_rss_failure(self, event): + FullPlugin.seen.append(event) + + @hook(points.MESSAGE_TEMPLATE) + def template(self, message, event, channel): + return RenderedMessage(f"{self.config.greeting}: {message.title}", message.body) + + +MANIFEST = """ +[plugin] +id = "full" +name = "Full" +version = "1.0.0" +sdk = ">=0.3,<1" +entry = "x:Y" +""" + + +async def test_plugin_end_to_end(registry, monkeypatch, tmp_path): + from module.core import AppContext + from module.core.scheduler import Scheduler + from module.mcp.tools import all_tools, handle_tool + + # AppContext 会设置进程级 bus/runner；用 monkeypatch 让测试结束后复原 + monkeypatch.setattr(host, "_bus", None) + monkeypatch.setattr(host, "_runner", None) + candidate = PluginCandidate( + manifest=parse_manifest(MANIFEST), + source="builtin", + load=lambda: FullPlugin, # type: ignore[arg-type] + ) + settings_obj = SimpleNamespace( + plugins=Plugins(options={"full": {"greeting": "yo"}}) + ) + manager = PluginManager( + settings_obj, + registry=registry, + discover_fn=lambda: ([candidate], []), + data_root=tmp_path, + ) + notifier = NotificationManager(load_providers=False) + ctx = AppContext( + settings_obj, notifier, Scheduler([]), MagicMock(), plugins=manager + ) + assert host.get_bus() is manager.bus + assert host.get_runner() is manager.runner + FullPlugin.seen = [] + try: + with patch("module.core.context.reset_llm_parser"): + await manager.start() + assert manager.statuses()[0].state == "active" + + # 事件：send_event → 总线 → 插件订阅者 + event = RssFailureEvent(rss_name="Feed", rss_url="http://x", error="boom") + with patch("module.notification.manager.record_event", new_callable=AsyncMock): + await notifier.send_event(event) + await manager.bus.drain() + assert FullPlugin.seen == [event] + + # 模板 + title, _ = await render_event(event, "bark") + assert title == f"yo: {event.describe()[0]}" + + # MCP + assert "full__greet" in [t.name for t in all_tools()] + [content] = await handle_tool("full__greet", {}) + assert json.loads(content.text) == {"greeting": "yo"} + + # 路由 + app = FastAPI() + app.include_router(v1, prefix="/api") + app.dependency_overrides[get_context] = lambda: ctx + + async def user(): + return "u" + + app.dependency_overrides[get_current_user] = user + client = TestClient(app) + assert client.get("/api/v1/plugins/full/greet").json() == {"greeting": "yo"} + + # 停用后全部消失 + settings_obj.plugins.enabled["full"] = False + await manager.apply_settings() + assert client.get("/api/v1/plugins/full/greet").status_code == 404 + assert "full__greet" not in [t.name for t in all_tools()] + assert await render_event(event, "bark") == event.describe() + finally: + await manager.stop() + + +# ---------------------------------------------------------------- MCP SSE + + +async def test_mcp_sse_endpoint_sends_response_only_once(monkeypatch): + """/sse 结束后不能再发第二次 http.response.start（3.x 起每次断开都报 + AssertionError）。""" + from contextlib import asynccontextmanager + + from module.mcp import server as mcp_server + + @asynccontextmanager + async def fake_connect_sse(scope, receive, send): + await send({"type": "http.response.start", "status": 200, "headers": []}) + await send({"type": "http.response.body", "body": b"", "more_body": False}) + yield (object(), object()) + + async def fake_run(*args, **kwargs): + return None + + monkeypatch.setattr(mcp_server.sse, "connect_sse", fake_connect_sse) + monkeypatch.setattr(mcp_server.server, "run", fake_run) + sent: list[dict] = [] + + async def receive(): + return {"type": "http.disconnect"} + + async def send(message): + sent.append(message) + + await mcp_server.handle_sse( + {"type": "http", "method": "GET", "path": "/sse", "headers": []}, + receive, + send, + ) + assert [m["type"] for m in sent].count("http.response.start") == 1 diff --git a/docs/dev/plugins.md b/docs/dev/plugins.md index 3a257c1f..6170eb09 100644 --- a/docs/dev/plugins.md +++ b/docs/dev/plugins.md @@ -14,6 +14,9 @@ | 搜索站点 | `points.SEARCH_SITE` | `SearchSite` | 搜索框的站点列表 | | 定时任务 | `points.SCHEDULED_TASK` | `ScheduledTask` | 插件启用即生效 | | 元数据源 | `points.METADATA_PROVIDER` | `MetadataProvider` | RSS 订阅的「解析器」 | +| REST 路由 | `points.API_ROUTER` | `fastapi.APIRouter` | 插件启用即生效 | +| MCP 工具 | `points.MCP_TOOL` | `McpTool` | 插件启用即生效 | +| MCP 资源 | `points.MCP_RESOURCE` | `McpResource` | 插件启用即生效 | 以及以下钩子（`@hook`，插件启用即生效）： @@ -23,8 +26,9 @@ | `title.parsed` | `points.TITLE_PARSED` | transform | 修正标题解析结果 | | `torrent.adding` | `points.TORRENT_ADDING` | transform | 修改发给下载器的保存路径、分类、标签 | | `http.request` | `points.HTTP_REQUEST` | transform | 修改 AB 发出的 GET 请求头（私有站 Cookie 等） | +| `message_template` | `points.MESSAGE_TEMPLATE` | transform | 改写系统事件推送到外部渠道的文案 | -此外，插件可以订阅事件（`@subscribe`），并使用私有的键值存储和数据目录。重命名等整理流水线的扩展点会在后续版本开放。 +此外，插件可以订阅系统事件（`@subscribe`），并使用私有的键值存储和数据目录。重命名等整理流水线的扩展点会在后续版本开放。 ## 目录结构 @@ -296,7 +300,112 @@ class MyPlugin(Plugin): return BangumiTv() ``` -`request.kind` 区分番剧与电影，`request.torrent.homepage` 是种子的详情页。返回值是完整的新元数据，通常基于 `request.current` 修改。失败或超时时保留原值，并计入熔断。目前 WebUI 的订阅表单只提供内置取值，插件元数据源需要通过 API 设置订阅的 `parser` 字段。 +`request.kind` 区分番剧与电影，`request.torrent.homepage` 是种子的详情页。返回值是完整的新元数据，通常基于 `request.current` 修改。失败或超时时保留原值，并计入熔断。添加订阅时，「解析器」下拉框会列出插件提供的元数据源。 + +### REST 路由 + +```python +from fastapi import APIRouter + +@provider(points.API_ROUTER, id="api") +def api(self): + router = APIRouter() + + @router.get("/stats") + async def stats(): + return {"count": await self.ctx.kv.get("count", 0)} + + return router +``` + +- 路由挂载在 `/api/v1/plugins/<插件 id>/` 下，上例即 `GET /api/v1/plugins/my-plugin/stats`。 +- **所有插件路由都强制登录鉴权**，与 WebUI 其它接口使用同一套凭据（会话 Cookie 或 `scope=api` 的 API 令牌）。插件无法注册匿名端点；未登录返回 401。 +- 路由随插件启用出现、随停用消失，无需重启。同一插件可以提供多个 `API_ROUTER`，它们会合并到同一前缀下。 +- 处理函数抛出 `HTTPException` 正常返回对应状态码；抛出其它异常返回 500，并计入熔断。 +- 插件路由不出现在 `/docs` 的 OpenAPI 文档中。 + +### MCP 工具与资源 + +```python +from ab_sdk.mcp import McpResource, McpTool + +@provider(points.MCP_TOOL, id="search") +def search_tool(self): + async def handler(args: dict): + return {"results": [...], "keyword": args["keyword"]} + + return McpTool( + description="在私有站点搜索", + handler=handler, + input_schema={ + "type": "object", + "properties": {"keyword": {"type": "string"}}, + "required": ["keyword"], + }, + ) + +@provider(points.MCP_RESOURCE, id="stats") +def stats_resource(self): + async def handler(): + return {"count": 3} + + return McpResource(name="统计", handler=handler) +``` + +- 对外名称自动加插件 id 前缀，避免冲突：工具名为 `<插件 id>__`（双下划线），资源 URI 为 `autobangumi://plugins/<插件 id>/`。不用 `.` 分隔，是因为不少 MCP 客户端会把工具名原样交给 LLM API，而后者通常只接受 `^[a-zA-Z0-9_-]{1,64}$`；不合规的工具名会被跳过并记录日志。 +- 工具处理函数返回可 JSON 序列化的对象；资源处理函数返回字符串（原样返回）或可 JSON 序列化的对象。抛出的异常以 `{"error": "..."}` 返回给客户端，并计入熔断。工具默认超时 60 秒（`McpTool(timeout=...)`）。 +- MCP 端点沿用 AB 的访问控制（IP 白名单或 `scope=mcp` 令牌），插件无需自行鉴权。 + +### 订阅系统事件 + +```python +from ab_sdk import subscribe +from ab_sdk.events import RssFailureEvent + +@subscribe("rss_failure") +async def on_rss_failure(self, event: RssFailureEvent): + self.ctx.log.warning("订阅 %s 失败：%s", event.rss_name, event.error) +``` + +AB 发出的、会进入通知中心的事件都是 `ab_sdk.events.SystemEvent` 的子类，带有 `severity`、`payload()`、`describe()`（默认中文标题与正文）和 `i18n()`（前端 i18n key 与参数）。它们先写入通知中心，再发布到事件总线，最后推送到外部通知渠道；关闭「通知」开关只影响外部推送，不影响插件收到事件。 + +| kind | 事件类 | 何时发布 | +| --- | --- | --- | +| `rss_failure` | `RssFailureEvent` | RSS 订阅从正常变为连接异常 | +| `download_failure` | `DownloadFailureEvent` | 种子重试后仍添加失败 | +| `offset_review` | `OffsetReviewEvent` | 番剧的季度 / 集数偏移需要人工确认 | +| `downloader_unavailable` | `DownloaderUnavailableEvent` | 下载器连不上、凭据错误或 IP 被封 | +| `update_available` | `UpdateAvailableEvent` | 检查到新版本 | +| `update_applied` / `update_failed` | `UpdateAppliedEvent` | 在线更新成功 / 失败（`kind` 随 `success` 变化） | +| `llm_auth_failure` | `LLMAuthFailureEvent` | 订阅类 LLM 提供商凭据失效 | +| `llm_plugin_install_failed` | `LLMPluginInstallFailedEvent` | LLM 插件安装失败 | +| `rename_conflict` | `RenameConflictEvent` | 媒体文件重命名遇到目标路径冲突 | +| `plugin.loaded` | `PluginLoaded` | 插件加载成功 | +| `plugin.disabled` | `PluginDisabled` | 插件加载失败或被熔断 | +| `inbox.changed` | — | 通知中心有新消息、已读或删除（宿主内部使用，供 SSE 推送） | + +系统事件的 `kind` 沿用 3.x 的取值（通知中心按它存储和翻译），因此不带 `.` 前缀；插件自己的事件仍必须以 `<插件 id>.` 开头。后续版本会陆续增加流水线事件（如 `torrent.added`、`file.renamed`）。 + +### 通知文案模板 + +系统事件推送到外部渠道前，会依次经过 `message_template` 钩子。钩子拿到默认文案、事件和渠道类型，返回新的 `RenderedMessage` 即可改写，返回 `None` 表示不修改： + +```python +from ab_sdk import hook, points +from ab_sdk.notify import RenderedMessage + +@hook(points.MESSAGE_TEMPLATE) +def template(self, message: RenderedMessage, event, channel: str): + if event.kind != "rss_failure": + return None + if channel == "telegram": + return RenderedMessage(f"[告警] {message.title}", f"{event.rss_name}\n{event.error}") + return None +``` + +- 只影响外部推送；通知中心里的文案由前端按 `i18n()` 渲染，不受影响。 +- 钩子对每个启用的渠道各调用一次。钩子出错或超时时沿用上一步的文案，并计入熔断；返回值类型不对时整条消息回退到默认文案。 +- 「新集数」通知暂不经过该钩子，仍使用通知渠道里配置的单集模板。 ## 测试 diff --git a/docs/plans/2026-10-08-plugin-architecture-4.0-design.md b/docs/plans/2026-10-08-plugin-architecture-4.0-design.md index fb9894fb..df3fae4e 100644 --- a/docs/plans/2026-10-08-plugin-architecture-4.0-design.md +++ b/docs/plans/2026-10-08-plugin-architecture-4.0-design.md @@ -34,6 +34,17 @@ - **推迟**：`feed_source`（与 `FeedConfig.source` 替代 `RSSItem.parser` 一起设计）、`title_parser` 链、`admission_policy`、`bangumi_matcher`、`release_ranker`、`rule.created`、`save_path` Provider，以及第 11 节的批量 `accept_many`。 - `module/plugin/context.py` 改为延迟 import `module.database`，否则网络层引用 `module.plugin.host` 会形成循环依赖。 +### 实施中的调整（P5） + +- **系统事件类移入 `ab_sdk.events`**，`SystemEvent` 由封闭 Union 改为基类（插件也可定义可通知事件）；`module.notification.events` 只做再导出。`kind` 沿用 3.x 取值（`rss_failure` 等，通知中心按它存储与翻译），不改成 `rss.failure` 式的点分名。 +- **通知管理器仍是事件入口**：第 4.2 节设想的「通知管理器、inbox 都作为总线订阅者」未实施。`NotificationManager.send_event` 依次写通知中心、发布到总线、推送外部渠道，调用方与外部行为不变；改成纯总线驱动要等 P3/P4 的流水线事件稳定后再统一。 +- **SSE 只有 `notification` 改为订阅总线**（`inbox.changed`，通知中心写入 / 已读 / 删除时发布，立即推送）。status / downloader / log / update 是状态快照而非事件，仍按节拍采样；`inbox_revision` 计数保留，作为帧里的 `revision` 字段。 +- **插件路由用分发路由实现**：FastAPI 不能卸载路由，宿主只注册 `/api/v1/plugins/{plugin_id}/{path:path}`（鉴权依赖在此强制），按插件 id 转发给由插件 `APIRouter` 合并成的 ASGI 应用，插件变更时重建。插件路由不进 OpenAPI。 +- **按插件作用域的 Provider id**：`ExtensionPoint` 新增 `scoped`，`api_router` / `mcp_tool` / `mcp_resource` 的 id 只需在插件内唯一（注册表键为 `/`）。MCP 工具名用 `__` 而非 `.`，因为 Anthropic / OpenAI 等 LLM API 的工具名只接受 `^[a-zA-Z0-9_-]{1,64}$`。 +- **`message_template` 只作用于系统事件的外部推送**，钩子签名 `(RenderedMessage, event, channel)`，按渠道各调用一次；「新集数」通知仍用渠道里的单集模板（`_format_message` 是同步接口，改造留到通知渠道迁到 `config_model` 时）。 +- **i18n（第 11 节风险 3）先给出 key**：`SystemEvent.i18n()` 返回 `notifications.kind.` 与 `payload()`，与前端现有文案键一致；外部推送与通知中心的中文兜底文案不变。插件事件的前端翻译依赖 P6 的 `host.i18n`。 +- 进程级访问器 `module.plugin.host.get_bus/set_bus`、`get_runner/set_runner`、`publish()`，由 `AppContext` 构造时设置，未设置时为空操作。MCP 的 `tools/list_changed` 通知未实现（客户端重新 list 即可看到插件工具变化）。 + ## 1. 背景与目标 AB 目前只有 **LLM 提供商** 是真正的运行时插件系统：签名下载、目录加载、懒导入、热重载。 @@ -581,7 +592,7 @@ organize: downloader.completed → media_files.classify → file_parser | **P2.5 多下载器** | 下载器多实例；`downloader_id` 列与迁移；按实例路由 add / rename / delete；organize 逐实例扫描 | qb + aria2 并存的 e2e 用例；单实例行为不变 | | **P3 流水线插件化：ingest** | `torrent.filter`、`title.parsed`、`torrent.adding`、`http.request` 钩子；`metadata_provider`（mikan / tmdb 以 `core` 登记）；内置插件 `ingest-filters`（包含过滤） | 已完成；无插件时行为不变（全量测试）。`feed_source`、`title_parser` 链、`admission_policy`、`matcher`、`ranker`、`save_path`、size 过滤、按订阅覆盖推迟，见第 0 节 | | **P4 流水线插件化：organize** | `media_files`、`file_parser`、`rename_strategy`（含 `template`）、`conflict_policy`、`file.renamed` 等事件 | 模板重命名；硬链接示例插件 | -| **P5 事件与外部接口** | SSE 改订阅 bus；`api_router`、`mcp_tool` 扩展点；`message_template` | 删掉 SSE 轮询；插件 MCP 工具 | +| **P5 事件与外部接口** | SSE 改订阅 bus；`api_router`、`mcp_tool` 扩展点；`message_template` | 已完成：系统事件上总线、通知中心 SSE 改为事件推送、插件路由 / MCP 工具与资源 / 通知模板；status 等快照类 SSE 仍按节拍采样。调整见第 0 节 | | **P6 前端插件** | Web Component 挂载点、`AbHost` 桥接、错误边界、`/plugins//web` 静态资源、`@autobangumi/plugin-ui` 包 | 示例插件「手动选种」以详情页标签形式可用 | | **P7 生态** | 插件管理页（安装、启停、日志、错误）、签名目录发布流程、模板仓库（含前端模板）、`ab-plugin` CLI、文档（中 / 英 / 日） | 6 个以上示例插件上架 | | **P8 发布** | beta 测试、性能对比（RSS 刷新耗时、内存）、升级指南、`docs/changelog/4.0.md` | `4.0.0-beta.1` → `4.0.0` |