feat(4.0-P5): 系统事件上总线、通知中心 SSE 改为事件推送、插件路由 / MCP / 通知模板

- 系统事件类移入 ab_sdk.events(SystemEvent 基类,新增 i18n()),
  NotificationManager.send_event 写入通知中心后发布到插件事件总线
- module.plugin.host 新增 get_bus/set_bus、get_runner/set_runner、publish,
  由 AppContext 构造时设置
- SSE notification 帧改为订阅 inbox.changed 即时推送,线格式不变
- 新扩展点 api_router(/api/v1/plugins/<id>/ 分发路由,强制鉴权)、
  mcp_tool / mcp_resource(<plugin-id>__<id> 命名)、message_template
- ExtensionPoint.scoped:Provider id 只需在插件内唯一
- ab_sdk 0.3.0;插件开发文档、设计文档、CHANGELOG 更新

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01EoBUPFw7Ty1epuPESdELs1
This commit is contained in:
Claude
2026-10-08 06:34:43 +00:00
parent 332fb451c4
commit 93f6d794c5
26 changed files with 1791 additions and 369 deletions

View File

@@ -4,6 +4,8 @@
- **新增(插件运行时)**:新增插件 SDK `ab_sdk`,包括 `Plugin` 基类、`@hook` / `@provider` / `@subscribe`、`PluginContext`,以及供插件作者写单元测试的 `ab_sdk.testing`;新增宿主运行时 `module/plugin`,负责清单解析、内置 / 本地目录 / pip entry point 三种来源的发现与加载、扩展点注册表、钩子超时与熔断、事件总线和插件私有 KV 存储(数据库迁移 v25)。新增 `plugins` 配置段和 `GET /api/v1/plugins`。本地与 pip 插件未签名,需开启 `plugins.allow_unsigned` 才会加载
- **新增(插件扩展点)**:插件可以提供下载器、通知渠道、LLM 解析提供商、搜索站点和定时任务(`ab_sdk.points`)。内置的 qB / aria2 和各通知渠道走同一套扩展注册表。设置页新增「插件」卡片,可查看状态、启停插件,并通过插件配置模型自动生成的表单修改配置;`secret_field` 声明的秘密字段在读取接口中一律掩码。下载器类型和通知渠道下拉框会列出插件提供的选项。新增插件开发文档(开发者指南 → 插件开发)
- **新增(事件与外部接口)**:系统事件(RSS 订阅异常、种子添加失败、下载器不可用、新版本等)在写入通知中心后发布到插件事件总线,插件可用 `@subscribe("rss_failure")` 等订阅;事件类移入 `ab_sdk.events`(`SystemEvent` 基类,新增 `i18n()` 返回前端 i18n key 与参数)。插件可以提供 REST 路由(挂载于 `/api/v1/plugins/<id>/`,强制登录鉴权,随插件启停出现和消失)、MCP 工具(工具名 `<插件 id>__<id>`)与 MCP 资源(`autobangumi://plugins/<插件 id>/<id>`),并可通过 `message_template` 钩子按事件类型和渠道改写系统通知的标题与正文。`ab_sdk` 升至 0.3.0
- **优化(通知中心推送)**:SSE 的 `notification` 帧改为订阅事件总线,通知中心有新消息、已读或删除时立即推送(原先每 3 秒比较一次修订号),推送格式不变
- **修复(搜索)**:`GET /api/v1/search/provider` 读取的是导入时的站点快照,新保存的搜索站点要重启后才会出现在列表中;现在实时读取
- **破坏性变更(升级路径)**:只支持从 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

View File

@@ -12,7 +12,7 @@ from .events import Event, PluginDisabled, PluginLoaded
from .hooks import Verdict, hook, provider, subscribe
from .plugin import Plugin
SDK_VERSION = "0.2.0"
SDK_VERSION = "0.3.0"
__all__ = [
"SDK_VERSION",

View File

@@ -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.<kind>``,其下有 ``title`` / ``body`` 两条文案;
- ``severity`` 为 ``info`` / ``warning`` / ``error``;
- ``once=True`` 表示同 ``dedup_key()`` 终生只入库一次(如「新版本可用」)。
插件也可以继承它定义自己的可通知事件(``kind`` 须以 ``<plugin-id>.`` 开头)。
"""
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.<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}",
)

52
backend/src/ab_sdk/mcp.py Normal file
View File

@@ -0,0 +1,52 @@
"""MCP 工具与资源契约。
插件用 ``@provider(points.MCP_TOOL, id="search")`` 返回 :class:`McpTool`,
用 ``@provider(points.MCP_RESOURCE, id="stats")`` 返回 :class:`McpResource`。
``id`` 只需在插件内唯一,宿主对外暴露时加上插件 id 前缀以避免冲突:
- 工具名:``<plugin-id>__<id>``(双下划线。不用 ``.``,因为不少 MCP 客户端
所接的 LLM API 只接受 ``^[a-zA-Z0-9_-]{1,64}$`` 的工具名)
- 资源 URI:``autobangumi://plugins/<plugin-id>/<id>``
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"

View File

@@ -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

View File

@@ -16,3 +16,25 @@ NOTIFIER = "notifier"
LLM_PROVIDER = "llm_provider"
SEARCH_SITE = "search_site"
SCHEDULED_TASK = "scheduled_task"
# --- P5:外部接口与通知模板 ---
#
# Provider 扩展点(id 只需在插件内唯一,宿主对外暴露时加插件 id 前缀):
#
# - ``API_ROUTER``:返回 ``fastapi.APIRouter``,挂载于
# ``/api/v1/plugins/<plugin-id>/``,强制登录鉴权
# - ``MCP_TOOL``:返回 :class:`ab_sdk.mcp.McpTool`,MCP 工具名为
# ``<plugin-id>__<id>``
# - ``MCP_RESOURCE``:返回 :class:`ab_sdk.mcp.McpResource`,URI 为
# ``autobangumi://plugins/<plugin-id>/<id>``
#
# 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"

View File

@@ -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)

View File

@@ -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))

View File

@@ -0,0 +1,69 @@
"""插件 REST 路由的分发端点:``/api/v1/plugins/{plugin_id}/{path:path}``。
鉴权在这里以路由依赖的形式强制(``get_current_user``),插件路由无法绕过,
也就无法注册匿名端点。请求转发给 ``AppContext.plugin_routes`` 中该插件的
路由;插件未启用或未提供路由时返回 404。
该路由至少需要 ``/plugins/<id>/`` 之后的一段(可为空),因此不会遮蔽
``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/<id> 之后的部分)
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)

View File

@@ -20,7 +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.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 (
@@ -37,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
@@ -89,7 +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
# 宿主各处(通知管理器、站内通知中心、流水线)经 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
@@ -108,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()

View File

@@ -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/<id>`` 之后的路径。插件处理函数抛出的
异常计入熔断后继续向上抛,由宿主的异常中间件返回 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

View File

@@ -0,0 +1,163 @@
"""插件提供的 MCP 工具与资源(points.MCP_TOOL / points.MCP_RESOURCE)。
每次 list / call 都从扩展注册表现取,插件加载、重载、停用后立即生效。对外
名称加插件 id 前缀以避免与内置及其它插件冲突:
- 工具名 ``<plugin-id>__<id>``:不少 MCP 客户端会把工具名原样交给 LLM API,
而 Anthropic / OpenAI 的工具名只接受 ``^[a-zA-Z0-9_-]{1,64}$``,所以不用 ``.``。
插件 id 只含小写字母、数字与连字符,第一个 ``__`` 即为分隔符,不会歧义。
- 资源 URI ``autobangumi://plugins/<plugin-id>/<id>``。
插件处理函数失败或超时会计入该插件的熔断计数。
"""
import asyncio
import logging
import re
from typing import Any
from mcp import types
from pydantic import AnyUrl
from ab_sdk import points
from ab_sdk.mcp import 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) -> list[tuple[ProviderEntry, Any]]:
"""调用该扩展点的全部插件工厂,跳过失败或类型不符的条目。"""
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)}")
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))
return resolved
# ------------------------------------------------------------------ tools
def tool_name(plugin_id: str, tool_id: str) -> str:
return f"{plugin_id}{TOOL_SEPARATOR}{tool_id}"
def _plugin_tools() -> dict[str, tuple[ProviderEntry, McpTool]]:
tools = {}
for entry, tool in _resolve(points.MCP_TOOL, McpTool):
name = tool_name(entry.plugin_id, entry.id)
if not _TOOL_NAME.match(name):
logger.warning(
"[Plugin:%s] MCP 工具名 %r 不合法(仅限字母、数字、_、-,"
"不超过 64 字符),已跳过",
entry.plugin_id,
name,
)
continue
tools[name] = (entry, tool)
return tools
def list_plugin_tools() -> list[types.Tool]:
return [
types.Tool(
name=name, description=tool.description, inputSchema=tool.input_schema
)
for name, (_, tool) 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 _plugin_resources() -> dict[str, tuple[ProviderEntry, McpResource]]:
return {
resource_uri(entry.plugin_id, entry.id): (entry, resource)
for entry, resource in _resolve(points.MCP_RESOURCE, McpResource)
}
def list_plugin_resources() -> list[types.Resource]:
return [
types.Resource(
uri=AnyUrl(uri),
name=resource.name,
description=resource.description or None,
mimeType=resource.mime_type,
)
for uri, (_, resource) 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
try:
result = await resource.handler()
except Exception as e:
_record(entry.plugin_id, f"MCP 资源 {entry.id}: {type(e).__name__}: {e}")
raise
_record(entry.plugin_id, None)
return True, result

View File

@@ -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/<id>/...``)。"""
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/<plugin-id>/<id>`` - 插件提供的资源
"""
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}"})

View File

@@ -18,10 +18,10 @@ from starlette.requests import Request
from starlette.responses import Response
from starlette.routing import Mount, Route
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()

View File

@@ -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]:
"""内置工具 + 当前已启用插件提供的工具(``<plugin-id>__<id>``)。"""
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]:

View File

@@ -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

View File

@@ -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",
]

View File

@@ -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:

View File

@@ -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

View File

@@ -0,0 +1,40 @@
"""系统事件推送文案:默认 ``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()`` 完全一致;
钩子失败或超时沿用上一步的结果(并计入插件熔断),返回值类型不对时整体
回退到默认文案。
"""
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
)
except Exception:
# 单个钩子的失败已在 runner 内隔离;这里兜底的是运行时本身的异常
# (如扩展点未声明),任何情况下都不能让通知因模板而发不出去
logger.warning("[Notification] 渲染通知模板失败,使用默认文案", exc_info=True)
return title, body
if not isinstance(message, RenderedMessage):
logger.warning(
"[Notification] message_template 钩子返回了 %s 而非 RenderedMessage,"
"使用默认文案",
type(message).__name__,
)
return title, body
return message.title, message.body

View File

@@ -8,12 +8,17 @@ AB 自己的下载器、通知渠道与搜索站点和第三方插件走同一
任何上下文里都能解析 Provider。
"""
import logging
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
logger = logging.getLogger(__name__)
CORE = "core"
@@ -25,9 +30,23 @@ POINTS = (
ExtensionPoint(points.LLM_PROVIDER, "provider", "LLM 解析提供商(llm.provider)"),
ExtensionPoint(points.SEARCH_SITE, "provider", "搜索站点"),
ExtensionPoint(points.SCHEDULED_TASK, "provider", "定时任务"),
# --- P5 events/api ---
ExtensionPoint(
points.API_ROUTER,
"provider",
"插件 REST 路由(/api/v1/plugins/<id>/)",
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 在构造时设置(即 PluginManager
# 持有的那一份);未设置时(单元测试、脚本)发布事件与执行钩子均为空操作
_bus: EventBus | None = None
_runner: HookRunner | None = None
def get_registry() -> ExtensionRegistry:
@@ -88,3 +107,39 @@ def _register_core(registry: ExtensionRegistry) -> None:
for provider_id, provider_cls in PROVIDER_REGISTRY.items():
_core(registry, points.NOTIFIER, provider_id, provider_cls)
# --- P5 events/api ---------------------------------------------------------
def get_bus() -> EventBus | None:
return _bus
def set_bus(bus: EventBus | None) -> None:
global _bus
_bus = bus
def get_runner() -> HookRunner | None:
return _runner
def set_runner(runner: HookRunner | None) -> None:
global _runner
_runner = runner
def publish(event: Event) -> None:
"""把宿主事件发布到进程级总线;总线未设置时什么也不做。
发布只负责入队,订阅者的失败不会影响调用方。没有运行中的事件循环时
(同步上下文)无法投递,记录日志后忽略。
"""
bus = _bus
if bus is None:
return
try:
bus.publish(event)
except RuntimeError:
logger.debug("[EventBus] 无事件循环,丢弃事件 %s", event.kind)

View File

@@ -18,6 +18,9 @@ class ExtensionPoint:
description: str = ""
# 仅 filter:钩子出错/超时时是否放行(True)还是视为拒绝(False)
fail_open: bool = True
# 仅 provider:True 时 Provider id 只需在插件内唯一(宿主对外暴露时会加
# 插件 id 前缀,如插件路由、MCP 工具),注册表按 ``<plugin_id>/<id>`` 存储
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():
@@ -96,5 +100,6 @@ class ExtensionRegistry:
)
def providers(self, point_name: str) -> dict[str, ProviderEntry]:
"""按 Provider id 索引;``scoped`` 扩展点的键为 ``<plugin_id>/<id>``。"""
self.point(point_name)
return dict(self._providers.get(point_name, {}))

View File

@@ -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

View File

@@ -0,0 +1,566 @@
"""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_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"
}
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()

View File

@@ -13,8 +13,11 @@
| LLM 解析提供商 | `points.LLM_PROVIDER` | `LLMProviderAdapter` 子类 | 设置 → LLM → 提供商 |
| 搜索站点 | `points.SEARCH_SITE` | `SearchSite` | 搜索框的站点列表 |
| 定时任务 | `points.SCHEDULED_TASK` | `ScheduledTask` | 插件启用即生效 |
| REST 路由 | `points.API_ROUTER` | `fastapi.APIRouter` | 插件启用即生效 |
| MCP 工具 | `points.MCP_TOOL` | `McpTool` | 插件启用即生效 |
| MCP 资源 | `points.MCP_RESOURCE` | `McpResource` | 插件启用即生效 |
此外,插件可以订阅事件(`@subscribe`),并使用私有的键值存储和数据目录。RSS 过滤、重命名等流水线扩展点会在后续版本陆续开放。
此外,插件可以订阅系统事件(`@subscribe`)、改写通知文案(`@hook(points.MESSAGE_TEMPLATE)`),并使用私有的键值存储和数据目录。RSS 过滤、重命名等流水线扩展点会在后续版本陆续开放。
## 目录结构
@@ -173,6 +176,111 @@ def sync(self):
任务与 RSS 刷新等内置任务一同启停,失败计入熔断。
### 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>__<id>`(双下划线),资源 URI 为 `autobangumi://plugins/<插件 id>/<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"<b>{event.rss_name}</b>\n{event.error}")
return None
```
- 只影响外部推送;通知中心里的文案由前端按 `i18n()` 渲染,不受影响。
- 钩子对每个启用的渠道各调用一次。钩子出错或超时时沿用上一步的文案,并计入熔断;返回值类型不对时整条消息回退到默认文案。
- 「新集数」通知暂不经过该钩子,仍使用通知渠道里配置的单集模板。
## 测试
`ab_sdk.testing` 可以在不启动 AB 的情况下构造并驱动插件:

View File

@@ -21,6 +21,17 @@
- **扩展点改名**:`search_provider` 改为 `search_site`,以区分用户在搜索设置里维护的站点列表。
- 签名目录来源(LLM 安装器泛化)、`dev_mode` 文件监听、bark / wecom 旧字段别名清理,移到 P7(生态)。
### 实施中的调整(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 只需在插件内唯一(注册表键为 `<plugin_id>/<id>`)。MCP 工具名用 `<plugin-id>__<id>` 而非 `.`,因为 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.<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 提供商** 是真正的运行时插件系统:签名下载、目录加载、懒导入、热重载。
@@ -568,7 +579,7 @@ organize: downloader.completed → media_files.classify → file_parser
| **P2.5 多下载器** | 下载器多实例;`downloader_id` 列与迁移;按实例路由 add / rename / delete;organize 逐实例扫描 | qb + aria2 并存的 e2e 用例;单实例行为不变 |
| **P3 流水线插件化:ingest** | `feed_source`、`title_parser` 链、`admission_policy`、`matcher`、`torrent.filter`、`ranker`、`metadata_provider` 链、`save_path`、`torrent.adding` | 新增 include / size 过滤;私有站 headers |
| **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/<id>/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` |