Auto_Bangumi/docs/dev/plugins.md

插件开发

::: warning 预览版 SDK 插件 SDK ab_sdk 在 4.0 期间为 0.x 版本，接口可能有不兼容调整；4.1 起冻结 1.0，之后遵守语义化版本。清单里的 sdk 字段声明你依赖的版本范围，AB 会拒绝加载不兼容的插件。 :::

插件是一个 Python 包，可以给 AutoBangumi 增加：

扩展点	常量	返回值	用户如何启用
下载器	points.DOWNLOADER	DownloaderFactory	设置 → 下载器 → 类型
通知渠道	points.NOTIFIER	NotifierFactory	设置 → 通知 → 添加渠道
LLM 解析提供商	points.LLM_PROVIDER	LLMProviderAdapter 子类	设置 → LLM → 提供商
搜索站点	points.SEARCH_SITE	SearchSite	搜索框的站点列表
定时任务	points.SCHEDULED_TASK	ScheduledTask	插件启用即生效
元数据源	points.METADATA_PROVIDER	MetadataProvider	RSS 订阅的「解析器」
重命名方式	points.RENAME_STRATEGY	RenameStrategy	设置 → 番剧管理 → 重命名方式
REST 路由	points.API_ROUTER	fastapi.APIRouter	插件启用即生效
MCP 工具	points.MCP_TOOL	McpTool	插件启用即生效
MCP 资源	points.MCP_RESOURCE	McpResource	插件启用即生效

以及以下钩子（@hook，插件启用即生效）：

扩展点	常量	类型	用途
torrent.filter	points.TORRENT_FILTER	filter	决定已匹配规则的种子是否下载
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，如文件重命名后的 file.renamed），并使用私有的键值存储和数据目录。

目录结构

config/plugins/local/my-plugin/     # 目录名必须与清单中的 id 相同
├── plugin.toml
├── my_plugin.py                    # 也可以是包，插件内部可以用相对导入
└── vendor/                         # 可选：纯 Python 依赖（禁止 .so/.pyd）

# plugin.toml
[plugin]
id = "my-plugin"              # 小写字母、数字、连字符
name = "我的插件"
version = "0.1.0"
sdk = ">=0.3,<1"              # 依赖的 ab_sdk 版本范围
entry = "my_plugin:MyPlugin"  # 模块:Plugin 子类，相对插件目录
description = "一句话说明"
permissions = ["network"]     # 仅用于向用户展示，不做强制

本地插件未经签名。要让 AB 加载它，需要在 设置 → 插件 中先打开「允许未签名插件」，再启用该插件。插件与 AB 在同一进程中运行，拥有完整权限。

第一个插件：Bark 风格的推送渠道

from pydantic import BaseModel, Field

from ab_sdk import Plugin, points, provider, secret_field
from ab_sdk.notify import NotificationMessage, NotifierSettings


class Options(BaseModel):
    endpoint: str = Field("https://push.example.com", title="推送地址")
    key: str = secret_field(description="推送密钥")


class Pusher:
    def __init__(self, plugin: "MyPlugin", settings: NotifierSettings) -> None:
        self.plugin = plugin
        self.settings = settings

    async def send(self, message: NotificationMessage) -> bool:
        import httpx

        async with httpx.AsyncClient(timeout=10) as client:
            resp = await client.post(
                f"{self.plugin.config.endpoint}/{self.plugin.config.key}",
                json={"title": message.title, "body": message.body},
            )
        return resp.is_success

    async def test(self) -> tuple[bool, str]:
        ok = await self.send(NotificationMessage(kind="event", title="AB", body="测试"))
        return ok, "已发送" if ok else "发送失败"


class MyPlugin(Plugin[Options]):
    config_model = Options

    @provider(points.NOTIFIER, id="my-push")
    def push(self):
        return lambda settings: Pusher(self, settings)

启用插件后：

1. 设置 → 插件里会出现由 Options 自动生成的表单。secret_field 声明的字段以密码框显示，读取配置的接口只返回掩码。
2. 设置 → 通知 → 添加渠道时，类型下拉框会多出 my-push。

核心概念

Plugin 与配置

• 继承 Plugin[配置模型] 并设置 config_model。AB 会按模型校验用户配置，插件通过 self.config 拿到带类型的实例。在 WebUI 保存不合法的值会被直接拒绝（HTTP 422）。
• async setup() 在加载后调用一次，async teardown() 在卸载前调用一次。配置变更时，插件会先 teardown，再重新构造并 setup。
• self.ctx 提供：
  • log：带 plugin.<id> 前缀的 logger
  • kv：插件私有的持久化键值存储，值须可 JSON 序列化
  • data_dir：config/plugin-data/<id>/
  • bus：事件总线

三种扩展声明

装饰器	用途
@provider(point, id=...)	工厂方法，返回该扩展点约定的实现
@hook(point, priority=...)	挂到宿主声明的 filter / transform 扩展点
@subscribe(kind)	订阅事件，"*" 表示全部事件

扩展点名写错时，插件会加载失败并在插件列表里显示原因，不会被悄悄忽略。

隔离与熔断

• 钩子、事件处理和定时任务都有超时。
• 同一插件连续失败 5 次会被自动禁用，并在插件列表中显示原因。修改该插件的配置后会重新尝试加载。
• 插件发布的事件，kind 必须以 <插件 id>. 开头。

各扩展点

下载器

from ab_sdk.downloader import AddResult, DownloaderCapabilities, DownloaderConnection


class MyClient:
    capabilities = DownloaderCapabilities(
        can_query=False, can_rename=False, can_manage=False, can_rss_rules=False
    )

    def __init__(self, conn: DownloaderConnection) -> None:
        self.conn = conn  # 来自设置页「下载器」的 host / username / password / ssl

    async def auth(self, retry: int = 3) -> bool: ...
    async def logout(self) -> None: ...
    async def add_torrents(self, torrent_urls, torrent_files, save_path, category, tags=None) -> AddResult: ...


class MyPlugin(Plugin):
    @provider(points.DOWNLOADER, id="my-client")
    def client(self):
        return MyClient

至少实现 CoreDownloaderClient（认证与添加种子）。需要重命名、整理等功能时，实现完整的 DownloaderClient 并在 capabilities 中声明，AB 会跳过未声明的操作。

LLM 提供商

返回 ab_sdk.llm.LLMProviderAdapter 的子类，@provider 的 id 必须等于其 info.id。内置提供商的 id 不能被覆盖。

搜索站点

from ab_sdk.search import SearchSite

@provider(points.SEARCH_SITE, id="my-site")
def site(self):
    return SearchSite(url="https://example.com/rss?q=%s", parser="tmdb")

用户在搜索设置里配置的同名站点优先。

定时任务

from ab_sdk.tasks import ScheduledTask

@provider(points.SCHEDULED_TASK, id="sync")
def sync(self):
    async def run():
        ...
    return ScheduledTask(run, interval=lambda: self.config.interval, initial_delay=60)

任务与 RSS 刷新等内置任务一同启停，失败计入熔断。

种子过滤（torrent.filter）

RSS 刷新时，种子先按规则匹配，再经过规则自带的「排除过滤」，最后交给 torrent.filter 钩子。任一钩子拒绝，该种子就不下载，也不会关联到番剧（与排除过滤相同）。手动「收集」整季时同样生效。

from ab_sdk import Plugin, Verdict, hook, points
from ab_sdk.ingest import BangumiInfo, Release, TorrentInfo


class OnlyHevc(Plugin):
    @hook(points.TORRENT_FILTER, priority=50)
    def check(
        self, torrent: TorrentInfo, release: Release | None, bangumi: BangumiInfo
    ) -> Verdict:
        if "HEVC" in torrent.name or "x265" in torrent.name:
            return Verdict.ok()
        return Verdict.reject("不是 HEVC")

• release 是标题解析结果，无法解析时为 None。
• 钩子出错或超时按放行处理（fail-open），并计入熔断。
• 返回 True / False 也可以，但 Verdict.reject(reason) 的原因会写进日志。

内置插件「种子过滤」（ingest-filters）就是这样实现的：在 设置 → 插件 中填写「包含过滤」后，只有名称匹配的种子才会下载。

修正解析结果（title.parsed）

在确定性解析（以及启用时的 LLM 解析）之后、准入判定之前调用。收到冻结的解析结果，用 dataclasses.replace 返回修改后的副本；返回 None 表示不修改。返回其他类型会被忽略并计入熔断。

from dataclasses import replace

ALIASES = {"LoliHouse": "Lolihouse"}


class GroupAlias(Plugin):
    @hook(points.TITLE_PARSED)
    def fix(self, release):
        if release.group in ALIASES:
            return replace(release, group=ALIASES[release.group])
        return None

新规则的番剧名、季度、字幕组都来自这里的结果，所以修改会影响新建的规则。改成 PV、合集等类型会让该资源不被收录。

修改添加请求（torrent.adding）

在种子交给下载器之前调用，参数是 ab_sdk.ingest.AddRequest：

from dataclasses import replace


class Tagger(Plugin):
    @hook(points.TORRENT_ADDING)
    def tag(self, request):
        return replace(request, tags=(*request.tags, request.bangumi.official_title))

注意：

• 整理与重命名只处理 Bangumi 分类下的种子，修改 category 等于让 AB 不再整理这些种子。
• ab:<番剧 id> 标签用于重命名时定位番剧，钩子删掉它时 AB 会补回。
• 重命名会从保存路径的 <番剧名>/Season N 结构推断番剧与季度，修改 save_path 时请保留这一结构。

请求头（http.request）

AB 发出的 GET 请求（RSS、种子文件、站点页面）都会经过这个钩子，适合给私有站点加 Cookie 或换 User-Agent。只能修改请求头，url 与 method 的改动会被忽略。

from ab_sdk import secret_field


class Options(BaseModel):
    cookie: str = secret_field(description="站点 Cookie")


class PrivateSite(Plugin[Options]):
    config_model = Options

    @hook(points.HTTP_REQUEST)
    def auth(self, request):
        if "pt.example.com" not in request.url:
            return None
        return replace(request, headers={**request.headers, "Cookie": self.config.cookie})

这个钩子在每个请求上都会调用，请保持轻量，不要在里面发起网络请求。

钩子会看到 AB 发出的所有 GET 请求的 URL，其中可能包含其他服务的密钥（例如 TMDB 的 api_key 查询参数）。只给目标站点添加 Cookie 等凭据，不要记录或转发其他请求的 URL。

元数据源

新番规则入库前，AB 按 RSS 订阅的「解析器」取值选择元数据源，补全官方标题、季度、年份和海报。内置 mikan 与 tmdb，插件可以增加新的取值：

from dataclasses import replace

from ab_sdk.ingest import Metadata, MetadataRequest


class BangumiTv:
    async def enrich(self, request: MetadataRequest) -> Metadata | None:
        info = await search(request.current.official_title)  # 自己的实现
        if info is None:
            return None  # 不修改
        return replace(request.current, official_title=info.name, poster_link=info.cover)


class MyPlugin(Plugin):
    @provider(points.METADATA_PROVIDER, id="bangumi-tv")
    def bgm(self):
        return BangumiTv()

request.kind 区分番剧与电影，request.torrent.homepage 是种子的详情页。返回值是完整的新元数据，通常基于 request.current 修改。失败或超时时保留原值，并计入熔断。添加订阅时，「解析器」下拉框会列出插件提供的元数据源。

REST 路由

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 工具与资源

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 令牌），插件无需自行鉴权。

订阅系统事件

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 表示不修改：

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() 渲染，不受影响。
• 钩子对每个启用的渠道各调用一次。钩子出错或超时时沿用上一步的文案，并计入熔断；返回值类型不对时整条消息回退到默认文案。
• 「新集数」通知暂不经过该钩子，仍使用通知渠道里配置的单集模板。

重命名方式

需要 sdk = ">=0.3,<1"。

from ab_sdk import Plugin, points, provider
from ab_sdk.rename import RenameInput


class Kebab:
    def target_name(self, f: RenameInput) -> str:
        if f.episode_type == "movie":
            return f"{f.bangumi_name}{f.full_suffix}"
        return f"{f.bangumi_name}-s{f.season:02d}e{f.episode:02}{f.full_suffix}"


class MyPlugin(Plugin):
    @provider(points.RENAME_STRATEGY, id="kebab")
    def kebab(self):
        return Kebab()

用户把 设置 → 番剧管理 → 重命名方式 设为 kebab 后，AB 对种子里的每个媒体文件和字幕文件各调用一次 target_name：

• RenameInput 字段：title（从文件名解析的标题）、bangumi_name（番剧文件夹名）、season、episode、suffix（含点的扩展名）、kind（media / subtitle）、language（字幕语言）、episode_type（episode / movie / special）、original_path（种子内当前路径）、group。
• 剧集偏移已经应用到 episode，季度取自番剧文件夹（同样已含季度偏移），策略只负责拼名字。半集（如 12.5）保留小数。
• 返回种子内的新相对路径，包括扩展名。字幕可以直接拼 f.full_suffix，它会生成 .zh.ass 这样的后缀。返回 f.original_path 表示不改名。
• 抛出异常、返回空字符串、绝对路径或含 .. 的路径时，AB 记录错误并保持原文件名，不会中断整理流程。
• 内置的 pn、advance、none 以 core 身份登记，不能被覆盖。内置插件「模板重命名」提供 template，可以作为参考实现。

整理事件

AB 在重命名完成后向事件总线发布两个事件，定义在 ab_sdk.organize：

事件	kind	发布时机
FileRenamed	file.renamed	下载器中的一个文件（媒体或字幕）被实际改名后；文件名已符合目标、无需改名时不发布
TorrentOrganized	torrent.organized	一个种子的媒体文件全部就位、字幕也处理完、打上 ab:renamed 标签时；重命名方式为 none 时不发布

from ab_sdk import Plugin, subscribe
from ab_sdk.organize import FileRenamed


class MyPlugin(Plugin):
    @subscribe(FileRenamed.kind)
    async def on_renamed(self, event: FileRenamed) -> None:
        self.ctx.log.info("%s → %s", event.old_path, event.new_path)

• FileRenamed 字段：torrent_hash、bangumi_id（未关联到番剧时为 None）、official_title、season、episode（已含偏移）、old_path、new_path、save_path、file_kind（media / subtitle）。
• TorrentOrganized 字段：torrent_hash、torrent_name、bangumi_id、official_title、save_path、collection（多文件合集）。
• 路径都是下载器视角：save_path 是种子保存目录，old_path / new_path 是种子内的相对路径。AB 与下载器不在同一台机器，或两者的挂载路径不同时，插件要自己做路径映射。
• 事件处理在独立队列里异步执行，有超时，失败计入熔断，不会拖慢或中断重命名。耗时操作（如调用媒体服务器）最好自己合并、延迟执行。内置插件「媒体库刷新」（media-server-refresh）就是这样做的：收到第一个 file.renamed 后等待一段时间，再请求 Jellyfin / Emby / Plex 刷新一次。

测试

ab_sdk.testing 可以在不启动 AB 的情况下构造并驱动插件：

from ab_sdk.testing import create_plugin

async def test_push(tmp_path):
    plugin, ctx = create_plugin(MyPlugin, {"key": "k"}, data_dir=tmp_path)
    await plugin.setup()
    factory = plugin.push()
    ...

以 pip 包发布

在包的 pyproject.toml 中声明 entry point，并在顶层包内附带 plugin.toml：

[project.entry-points."autobangumi.plugins"]
my-plugin = "my_plugin:MyPlugin"

pip 安装的插件同样未签名，需要开启「允许未签名插件」。Docker 镜像无法 pip 安装，此时请使用本地目录，并把依赖放进 vendor/。