插件开发指南

Reelay 的主流程（监听 → 识别 → 转存整理）是固定骨架，所有可变能力都挂在三类扩展点上。加功能就是写一个插件目录，不用改内核。

两个可以直接运行的示例：

examples/plugins/webhook_notify/：Webhook 通知渠道（Provider）
examples/plugins/quality_gate/：拦截 CAM/TS 等低画质资源（Hook）

目录结构

每个插件一个目录，插件代码写在 __init__.py，插件用到的其他文件（规则、模板、默认数据等）都放在同一个目录里，用 Path(__file__).with_name(...) 读取。内置插件（reelay/plugins/）与外部插件（data/plugins/）组织方式相同：

plugins/
├── organizer/          # 媒体库整理
│   ├── __init__.py     # 插件代码
│   ├── settings.py     # 默认设置与校验（内置插件）
│   ├── naming.py       # 插件内部模块
│   ├── category.py
│   └── category.yaml   # 二级分类规则
└── my_plugin/
    ├── __init__.py
    └── rules.yaml

插件的设置也属于插件：外部插件用 ui_fields（带默认值）与 validate_config 声明（见下文「在管理台展示与校验配置」）；内置插件还可以在目录里放 settings.py，提供 DEFAULTS 与 normalize(section)，配置读取与保存时由内核按目录调用，默认值会写进配置供管理台编辑。

内置插件目录里的 *.yaml 随安装包一起发布（pyproject.toml 的 package-data）；用其他后缀时把它加进去。

选哪类扩展点

你想做的事	扩展点	例子
同一件事换一种实现	Provider（能力槽）	新消息源、新通知渠道、另一种解析器或网盘
在某个阶段前后插入处理，可能改数据或拦截	Hook（钩子）	补元数据、二次过滤、选目录、整理
某件事发生后旁路做点什么，不能挡主流程	Event（事件）	统计、日报、写外部索引、审计

加载方式

插件目录：把整个目录放进 data/plugins/（Docker 部署把 deploy/plugins 挂载到 /app/data/plugins），重启后自动发现；只有一个文件的简单插件也可以直接放 *.py。
独立发行包：在你的包里声明 entry point，pip install 后自动加载：
toml
```
[project.entry-points."reelay.plugins"]
my_plugin = "my_package.reelay_plugin"
```

外部扩展导入或实例化失败只会记录到管理台"插件"页，不影响内置能力。扩展以 Reelay 进程权限运行，只加载你信任的代码。

注册

python

from reelay.core.registry import register

@register(kind="provider", ptype="notify")   # ptype: source / parser / decision / storage / notify
class MyNotify: ...

@register(kind="hook", stage="after_parse", order=70)
class MyHook: ...

@register(kind="event", event="on_saved")
class MyListener: ...

装饰器只记录类，discover() 时用无参构造函数实例化。每个扩展需要一个全局唯一的 name，它同时是 Provider 的引用名和配置里 plugins.<name> 的键：开关是 plugins.<name>.enabled，扩展自己的设置也放在 plugins.<name> 下。

贯穿全程的 PipelineContext

一条消息对应一个 PipelineContext（reelay/core/contracts.py），每个阶段从数据库快照重建：

属性	含义
`raw`	原始消息：`source`、`msg_id`、`text`、`urls`（含超链接真实 URL）
`parsed`	识别结果 `ParsedResource`：标题、年份、类型、季集、类型 ID、国家、版本信息（`release`：字幕、音轨、HDR、片源等，见 `reelay/core/release.py`）、分享码等
`state` / `decision`	当前状态与决策（save / ask / skip）
`save_dir_alias` / `save_dir_cid`	目标目录
`source_rule`	合并了 `defaults` 的来源规则
`ctx.ns(name)`	扩展私有命名空间（`meta[name]`），随任务持久化
`ctx.log(msg)`	写入任务轨迹，管理台可见

Hook

python

class MyHook:
    name = "my_hook"
    stage = "after_parse"   # before_parse / after_parse / before_save / after_save
    order = 70              # 同一阶段内从小到大执行；内置：organizer 选目录 10（before_save）与整理 20（after_save），tmdb_enrich 50，share_probe 60

    async def run(self, ctx) -> bool:
        ...
        return True          # False 表示拦截

返回 False 与抛出异常的后果因阶段而异：

阶段	返回 False	抛出异常
`before_parse` / `after_parse`	任务结束为 skipped	记录日志后继续执行后续 Hook（失败隔离）
`before_save`	任务结束为 skipped	任务失败并按阶段重试（转存前置检查不能静默跳过）
`after_save`	任务转 need_manual	记录整理错误，任务转 need_manual

同一阶段里不要和内置插件取相同的 order，否则先后顺序取决于加载顺序。

after_save 运行时文件已经转存完成。要用到存储 Provider 某些可选方法的扩展，在 requires_storage 里声明方法名，如媒体库整理的 requires_storage = ("library",)：配置里有规则使用不提供这些方法的存储时，保存配置会被拒绝，「插件」页也会标出不支持的存储。只做附加产物（如生成媒体服务器索引文件）的 Hook 应自己捕获异常并返回 True，不要影响任务结果。

Event

python

class MyListener:
    name = "my_listener"
    event = "on_saved"   # on_message / on_identified / on_rejected / on_saved / on_failed / on_daily

    async def handle(self, ctx) -> None:
        ...

事件与任务状态在同一事务内写入 Outbox，异步投递：至少一次、单次 30 秒超时、失败退避重试（最多 8 次；系统告警与恢复通知 24 小时内持续重试）、同一任务内按生成顺序。
ctx 是事件发生时的快照，修改它不会影响任务。
on_daily 由内核每天在 schedule.daily_hour 发出一次，ctx.meta["system"] 里有日期和过去 24 小时汇总，适合日报、巡检类扩展；进程启动时会等停机期间补拉的消息解析完再发（最多推迟 15 分钟）。
一个类只能订阅一个事件；需要订阅多个事件时写多个类并使用相同的 name（共用一个开关），参考 reelay/plugins/episode_tracker/。

Provider 契约

ptype	需要实现	可选
`source`	`async listen(emit)`：长驻协程，每条消息 `await emit(RawMessage(...))`	`configure(targets)`、`bind_watermark(lookup)`、`health()`、`async close()`
`parser`	`async parse(ctx) -> ParsedResource`	—
`decision`	`async decide(ctx) -> Decision`	—
`storage`	`prepare` / `save` / `reconcile`（只读核验，绝不能重发写请求）/ `abandon`（放弃结果不明的转存，删除暂存目录并读回确认）	见下
`notify`	`async notify(ctx, event)`；event 为 identified / pending_approval / saved / transfer_uncertain / organize_failed / failed	`async notice(text, html=False, buttons=None)`：系统通知（告警、汇总、日报）；不接受 `html` 的渠道收到去掉标签的文字，不接受 `buttons` 的只收到正文

存储 Provider 的可选方法（reelay/core/contracts.py 的 StorageProvider）：

方法	用途
`discard(ctx)`	清理 `prepare` 建立、但因去重而没用上的空暂存目录
`reconcile_partial(ctx)`	只读核验，按已经到了的文件出回执（人工选「按已到的继续」）
`ready() -> bool`	凭证是否就绪；没有时视为就绪
`dir_path(cid)`	目录在网盘里的路径，管理台与审批卡片据此显示真实去向
`library()`	整理用的网盘接口：`iterdir` / `get_attr` / `mkdir` / `rename` / `move` / `delete`，ID 为 int
`inspect_share(parsed, describe=...)`	只读查询分享是否可用、有哪些季集（分享检查插件用）
`browse(cid, offset, limit)`	列出子目录（管理台选目录用）

save 的结果用异常区分：TransferRejected 表示网盘明确拒绝、确定没有转存，可以重发；TransferUncertain 表示请求可能已经执行，之后只能核验（拿得到网盘对这次请求的回应时放进它的 response，内核写进任务轨迹供排查）；StorageBusy(reason, until) 表示网盘暂时拒绝一切请求或推迟了这次请求、这次请求没有执行，任务到 until 再继续，不算失败。prepare、reconcile 与可选方法遇到同样的情况也抛 StorageBusy。

Provider 由配置选择：providers.source、providers.notify，以及 defaults/sources[] 中的 parser、decision、storage。

内置的三个 decision 逐层叠加：default 只按规则；subscription（默认）在其上让订阅剧集的新集自动放行；llm 再按自然语言偏好筛选。自定义决策器想保留订阅放行，可以继承 SubscriptionDecision。开启 subscriptions_only 时，内核会先核对有效订阅，未匹配就直接忽略，不调用决策器；正式识别与测试预览一致。决策器对 LLM 降级结果返回 save 时，内核仍会改为转人工。

调用 AI

扩展需要调用 AI 时用 reelay/core/ai.py，与内置能力共用客户端配置、退避重试与调用记录。继承 ai.UsesAI 后，self._client / self._model 在用到时按当前的配置与密钥取得（管理台改了立即生效，未配置时 _client 为 None）；测试里直接给它们赋值即可替换：

python

from reelay.core import ai

class MyListener(ai.UsesAI):
    async def handle(self, ctx):
        if self._client is None:
            return                                     # 按规则降级
        try:
            result, _ = await ai.complete_json(
                self._client, model=self._model, purpose="my_plugin", task_id=ctx.task_id,
                messages=[...], schema={...}, name="my_output", validate=my_validate)
        except ai.AIError:
            return                                     # 拒答、非法输出、网络错误：按规则降级

validate 必须做本地严格校验并在不合格时抛 ValueError（非法输出会重试一次）；
每次请求记入调用记录（purpose 会显示在管理台「统计 → AI 调用」）；
送给 AI 的消息正文先用 ai.scrub() 去掉链接与访问码；
AI 的输出只作提议：涉及写操作的决定交给规则校验或用户确认。

元数据补全约定

识别完成后，内置的 tmdb_enrich（after_parse，order 50）按标题与年份匹配 TMDB 条目并回填元数据。TMDB 的 ID 体系是 Reelay 的数据标准：类型 ID 就是 TMDB 类型 ID，命名中的 [tmdbid=…] 供 Jellyfin / Emby 刮削，海报按 TMDB 图片路径显示。下面的字段由内核和其他扩展直接读取；接入其他数据源（如豆瓣、Bangumi）时，请把结果对应到 TMDB 条目并写入同样的字段。

写入 ctx.parsed：

字段	内容	谁在读
`title` / `year`	规范标题与上映 / 首播年份	没有 TMDB ID 时的作品身份（规范片名 + 年份，见 `reelay/core/works.py`，不同来源的同一部作品靠它合并）、整理路径、缺集追更
`genre_ids`	TMDB 类型 ID 整数列表，整体替换 AI 的值	二级分类（`plugins/organizer/category.yaml`，如 16 = 动画）
`origin_country`	两位大写国家/地区代码；数据源没有时保留 AI 的值，不要清空	二级分类
`raw_fields["tmdbid"]`	TMDB 条目 ID（整数）	作品身份（入库去重、订阅与追更按它区分同名作品）、命名模板变量 `tmdbid`（默认模板写入 `[tmdbid=…]`）、缺集追更查询已播集数
`raw_fields["poster_path"]`	TMDB 图片路径，如 `/abc.jpg`，不是完整 URL	管理台海报（拼接 `https://image.tmdb.org/t/p/<尺寸>`）；没有时显示文字封面

写入 ctx.meta["tmdb"]：确定性置信度在全部 after_parse Hook 之后计算，按这里的结果判断（reelay/core/confidence.py）。

情况	写入	对置信度的影响
匹配成功	`matched: True`、`tmdbid`	计为通过的信号
查询成功但没有可信匹配	`matched: False`	封顶 0.45，标记「TMDB 未匹配」，转人工确认
没有查询（未配置凭证）	`matched: False`、`note: 原因`	中立，不影响分数
查询出错（网络、限流等）	`matched: False`、`error: 异常类型名`	中立，不影响分数

candidates、kind、score 等其余字段只供任务详情排查，内核不读取。

已有任务的补跑：after_parse Hook 可以实现 needs_refresh(ctx) -> bool。主进程启动时，内核对待审批和已入库任务逐条询问，返回 True 的补跑该 Hook（不重新调用 LLM）：待审批任务整体采用新结果并重算置信度；已入库任务的片名、路径与去重键已经落地，只在片名不变时补上 poster_path。补跑只更新快照，不改任务的更新时间，审批超时与入库统计不受影响。tmdb_enrich 对配置 key 之前识别（有 note）或上次查询出错（有 error）的任务返回 True，所以配好 TMDB 重启后，已有任务也会补上规范元数据和海报。

接入其他数据源有两种方式：

补充：order 大于 50，只在 TMDB 没有匹配时查询；命中后写入上表字段，并整体替换 meta["tmdb"]——残留的 note / error 会让结果仍按中立处理。
替代：在 plugins 中关闭 tmdb_enrich，由你的 Hook 按同样约定写入。

python

import asyncio

from reelay.core import runtime
from reelay.core.registry import register


@register(kind="hook", stage="after_parse", order=55)
class DoubanFallback:
    name = "douban_fallback"

    async def run(self, ctx) -> bool:
        p = ctx.parsed
        if not p or not p.is_media or ctx.meta.get("tmdb", {}).get("matched"):
            return True
        hit = await asyncio.to_thread(lookup_via_douban, p.title, p.year)  # 你的实现：经豆瓣找到对应的 TMDB 条目
        runtime.observe("douban", "connected")
        if hit:
            p.title, p.year = hit["title"], hit["year"]
            p.raw_fields["tmdbid"] = hit["tmdbid"]
            # 能拿到时同样回填 genre_ids、origin_country、raw_fields["poster_path"]
            ns = ctx.ns("tmdb")
            ns.clear()
            ns.update(matched=True, tmdbid=hit["tmdbid"], source="douban")
        return True

对应不到 TMDB 条目的结果不要只写自有 ID：去重、分类、命名与追更都以 TMDB 条目为准，这种情况按「没有可信匹配」处理。

读取配置与使用内核服务

python

from reelay.core import runtime

section = runtime.plugin_config("my_plugin")           # 本扩展的设置 plugins.my_plugin（不含 enabled）
language = runtime.get("tmdb.language")                # 点路径读取系统配置
runtime.services().notify("📦 备份完成", key="my_plugin")  # 发系统通知到所有通知渠道（持久化投递）
state = runtime.services().state("my_plugin")          # 按扩展隔离的持久化 KV，值需可 JSON 序列化
state.set("last_run", "2026-10-19")
runtime.observe("my_api", "failed", "TimeoutError")    # 记录一次外部调用结果（connected / failed）

runtime.observe 记录的是最近一次真实调用的结果，不会因一段时间没有调用而过期，可在 GET /api/health 的 runtime 中查看；内置的 LLM 解析与 TMDB 补全用它驱动管理台「总览 → 运行状态」里「AI 解析」「TMDB 元数据」两行。

runtime.get / runtime.plugin_config 读到的是任务接收时的配置（按任务记下的配置版本），配置变更只影响新任务；不在任务作用域里时读进程当前的有效配置。services() 在内核调用 Hook、Event、通知渠道，以及解析/决策/存储 Provider 的方法时可用（包括其中的 asyncio.to_thread）；消息源的 listen 是长驻协程，不在任务作用域内。

services() 提供的方法：

方法	内容
`notify(text, key=, html=, buttons=)`	发系统通知到所有通知渠道（持久化投递）
`state(name)`	按扩展隔离的持久化 KV（`get` / `set` / `delete` / `items`）
`storage(name)`	名为 name 的存储 Provider（如整理插件取 `library()`）
`query(capability, *args)`	调用其他扩展声明的能力，见下
`subscriptions()` / `finish_subscription(id, record)`	剧集订阅；把播完且全部入库的订阅标为已结束
`ignored_works()`	不再接收的作品
`stored_series()`	按（作品, 季）汇总的已入库集：`[{work, name, season, episodes}]`；某部作品某一季已入库哪些集用 `works.episodes(stored, works.identity(片名, 年份, tmdbid), 季)` 查
`task(task_id)` / `task_record(task_id)`	任务快照（`PipelineContext`）/ 任务行（状态、错误、更新时间等）
`title_media_types()` / `titles_since(since)`	片名 → 媒体类型；某个时间之后新收的片名
`recent_feedback(source=, kinds=, limit=)`	最近的人工裁决（批准、忽略、修正），正文已去掉链接与访问码

扩展之间、内核与扩展之间不互相导入。扩展在 provides（能力名 → 方法名）里声明它能回答什么，内核和其他扩展用 services().query(能力名, ...) 调用；扩展关闭或没有安装时返回 None，调用方按没有这项能力处理。内置的能力：

能力	谁提供	方法与返回
`task_explanation`	失败解释	`(task_id) -> dict \| None`：这条任务最近一次的原因与建议操作，任务详情显示
`episode_report`	追更提醒	`() -> dict`：最近一次缺集巡检结果，日报、媒体库与订阅页使用
`bot_action:<动作>`	追更提醒	`(action, digest, original) -> (正文, 按钮, 提示)`：控制 Bot 上这个插件自己的按钮

python

class MyReport:
    name = "my_report"
    provides = {"weekly_summary": "summary"}

    def summary(self) -> dict:
        ...

# 其他扩展里
summary = runtime.services().query("weekly_summary") or {}

同步阻塞调用（HTTP、SDK）请放进 asyncio.to_thread，不要阻塞事件循环。

在管理台展示与校验配置

python

class MyHook:
    name = "my_hook"
    ui_label = "我的扩展"
    ui_description = "一句话说明，显示在插件列表"
    ui_fields = [   # 自动生成配置表单，并在保存时按类型校验；path 相对 plugins.my_hook
        {"path": "threshold", "label": "阈值", "type": "number", "min": 0, "max": 1, "default": 0.5},
        {"path": "mode", "label": "模式", "type": "string", "choices": ["a", "b"]},
        {"path": "strict", "label": "严格模式", "type": "boolean"},
    ]
    default_enabled = False   # 未设置 plugins.my_hook.enabled 时默认关闭（适合有副作用的扩展）

    def validate_config(self, cfg: dict) -> None:
        """扩展启用时，配置保存与启动都会调用；抛 ValueError 拒绝无效配置。"""
        section = runtime.plugin_config(self.name, cfg)

对应的配置文件：

yaml

plugins:
  my_hook:
    enabled: true
    threshold: 0.7
    mode: a

default 是未填写时扩展实际使用的值，只用于在表单里提示。数组、对象等复杂配置在 validate_config 中自行校验，用户通过配置文件维护。扩展不要读环境变量：需要的地址、开关写成 ui_fields；需要的文件放在 reelay.core.paths.DATA 下。

插件动作

管理台需要向插件要数据或让插件算点什么（如预览命名模板、展示内置规则）时，插件把普通方法列进 ui_actions，内核统一提供 POST /api/extensions/<插件名>/actions/<动作名>：

python

class MyHook:
    ui_actions = ("preview",)

    def preview(self, payload: dict) -> dict:
        """payload 是请求体（JSON 对象，可以为空）；返回 JSON 对象。"""
        if not payload.get("text"):
            raise ValueError("text 不能为空")   # ValueError / TypeError → 422，消息原样显示给用户
        return {"result": payload["text"].upper()}

只有列进 ui_actions 的方法可以调用，插件的其他方法从接口上取不到；插件未安装或没有这个动作时返回 404。
鉴权由内核完成（与其他管理接口相同）；方法里可以用 runtime.get 读当前配置。
插件不需要依赖 FastAPI。方法在 Web 进程的事件循环里执行，要快；耗时的写成 async def。
插件目录里会列出每个插件的 actions。

内置插件可以有专用的界面，放在前端的 web/src/plugins/<插件名>/：设置编辑器由后端声明 ui_config_type，并登记到 web/src/plugins/index.ts 的 PLUGIN_EDITORS；只服务于这个插件的界面（如媒体库整理设置里的「预览整理效果」弹窗）也放在这里；组件需要后端数据时调用 api.extensionAction(插件名, 动作名, payload)。前端是整体构建后发布的，外部插件没有专用编辑器，用 ui_fields 生成表单。

测试

用 discover(plugin_dirs=[...]) 加载你的扩展并直接调用，参考 tests/test_examples.py：

python

registry = discover(plugin_dirs=["path/to/plugins"])
assert registry.load_errors == []
hook = next(h for h in registry.hooks["after_parse"] if h.name == "my_hook")
with runtime.scope({"plugins": {"my_hook": {"threshold": 0.5}}}):
    assert await hook.run(ctx) is True

插件开发指南 ​

目录结构 ​

选哪类扩展点 ​

加载方式 ​

注册 ​

贯穿全程的 PipelineContext ​

Hook ​

Event ​

Provider 契约 ​

调用 AI ​

元数据补全约定 ​

读取配置与使用内核服务 ​

在管理台展示与校验配置 ​

插件动作 ​

测试 ​

插件开发指南

目录结构

选哪类扩展点

加载方式

注册

贯穿全程的 PipelineContext

Hook

Event

Provider 契约

调用 AI

元数据补全约定

读取配置与使用内核服务

在管理台展示与校验配置

插件动作

测试