feat(4.0-P3): ingest 流水线扩展点与内置种子过滤插件

- 钩子:torrent.filter(排除过滤之后,fail-open)、title.parsed(准入判定前)、
  torrent.adding(save_path / category / tags,ab:<id> 标签强制保留)、
  http.request(GET 请求头)
- Provider:metadata_provider,mikan / tmdb 以 core 登记,按 RSSItem.parser 选择,
  行为与 3.x 一致;插件元数据源走超时与熔断
- SDK 0.3.0:新增 ab_sdk.ingest(冻结快照 TorrentInfo / BangumiInfo / AddRequest /
  HttpRequest / Metadata)
- host.get_runner()/hook_runner():AppContext 构建时设置 runner,无钩子时零开销
- 内置插件 ingest-filters:全局包含过滤,默认启用、留空不生效
- WebUI:添加订阅的解析器下拉框并入插件元数据源
- 文档:插件开发指南新增各扩展点;设计文档 P3 调整

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:31:23 +00:00
parent 332fb451c4
commit 9b7ef00c2e
27 changed files with 1418 additions and 84 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` 声明的秘密字段在读取接口中一律掩码。下载器类型和通知渠道下拉框会列出插件提供的选项。新增插件开发文档(开发者指南 → 插件开发)
- **新增(ingest 扩展点)**:RSS 下载流程开放 `torrent.filter`(种子过滤)、`title.parsed`(修正解析结果)、`torrent.adding`(修改保存路径 / 分类 / 标签)、`http.request`(为私有站点添加 Cookie 等请求头)四个钩子,以及 `metadata_provider` 元数据源扩展点:内置 mikan / tmdb 以 `core` 身份登记,插件可按 RSS 订阅的「解析器」取值提供新的元数据源,添加订阅时的解析器下拉框会列出插件选项。插件 SDK 升至 0.3.0,新增 `ab_sdk.ingest`。未安装插件时行为不变
- **新增(种子过滤)**:新增默认启用的内置插件「种子过滤」(`ingest-filters`)。在 设置 → 插件 中填写「包含过滤」后,只下载名称匹配任一表达式的种子(对所有订阅生效);留空时不过滤
- **修复(搜索)**:`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

@@ -0,0 +1,154 @@
"""ingest 流水线(RSS → 解析 → 匹配 → 过滤 → 元数据 → 添加种子)的插件契约。
钩子与 Provider 收到的都是只读快照(冻结 dataclass),不是数据库对象:
插件不能、也不应该直接修改宿主的 ORM 实例。需要修改时用
:func:`dataclasses.replace` 返回一个新副本。
扩展点一览(常量见 :mod:`ab_sdk.points`):
- ``torrent.filter``(filter):``accept(torrent, release, bangumi) -> Verdict``
- ``title.parsed``(transform):``fix(release) -> release``
- ``torrent.adding``(transform):``adjust(request: AddRequest) -> AddRequest``
- ``http.request``(transform):``adjust(request: HttpRequest) -> HttpRequest``
- ``metadata_provider``(Provider):返回 :class:`MetadataProvider`
"""
from collections.abc import Mapping
from dataclasses import dataclass
from typing import Literal, Protocol
@dataclass(frozen=True, slots=True)
class TorrentInfo:
"""RSS 中的一条种子。"""
name: str
url: str
homepage: str | None = None
rss_id: int | None = None
@dataclass(frozen=True, slots=True)
class BangumiInfo:
"""番剧规则(订阅)的只读视图。"""
id: int | None
official_title: str
title_raw: str
season: int
group_name: str | None = None
episode_type: str = "episode"
rss_link: str = ""
save_path: str | None = None
# 规则上的排除过滤(逗号分隔),宿主已在调用 torrent.filter 前应用过
filter: str = ""
class Release(Protocol):
"""标题解析结果(``ParsedRelease``)的只读视图。
宿主传入的是冻结 dataclass,``title.parsed`` 钩子用
``dataclasses.replace(release, group=...)`` 返回修改后的副本。
下列字段之外的属性(``media_type``、``release_kind``、``codecs`` 等)
同样可读,但在 SDK 1.0 之前不保证稳定。
"""
@property
def raw(self) -> str: ...
@property
def title_en(self) -> str | None: ...
@property
def title_zh(self) -> str | None: ...
@property
def title_jp(self) -> str | None: ...
@property
def group(self) -> str | None: ...
@property
def season(self) -> int | None: ...
@property
def episode(self) -> int | float | None: ...
@property
def resolution(self) -> str | None: ...
@property
def source(self) -> str | None: ...
@property
def subtitle(self) -> str | None: ...
@property
def year(self) -> int | None: ...
@dataclass(frozen=True, slots=True)
class AddRequest:
"""即将交给下载器的添加请求,``torrent.adding`` 钩子可返回修改后的副本。
注意:
- 整理/重命名只扫描 ``Bangumi`` 分类的种子;改掉 ``category`` 意味着
AB 不再整理这些种子。
- ``ab:<id>`` 标签用于重命名时查找番剧,钩子删掉它时宿主会补回。
- 重命名会从 ``save_path`` 的 ``<番剧名>/Season N`` 结构推断番剧与季度,
修改时请保留这一结构。
"""
bangumi: BangumiInfo
torrents: tuple[TorrentInfo, ...]
save_path: str
category: str
tags: tuple[str, ...] = ()
@dataclass(frozen=True, slots=True)
class HttpRequest:
"""宿主发出的 GET 请求(RSS、种子文件、站点页面)。
``http.request`` 钩子可返回修改了 ``headers`` 的副本,例如为私有站点
添加 Cookie 或替换 User-Agent::
return replace(req, headers={**req.headers, "Cookie": "uid=1"})
"""
method: str
url: str
headers: Mapping[str, str]
@dataclass(frozen=True, slots=True)
class Metadata:
"""番剧/电影的展示元数据。"""
official_title: str
season: int = 1
year: str | None = None
poster_link: str | None = None
@dataclass(frozen=True, slots=True)
class MetadataRequest:
"""新规则入库前的元数据补全请求。
``current`` 是解析器得出的当前值。Provider 返回完整的新
:class:`Metadata`(通常是 ``replace(request.current, ...)``),
返回 None 表示不修改。
"""
kind: Literal["bangumi", "movie"]
torrent: TorrentInfo
language: str
episode_type: str
current: Metadata
class MetadataProvider(Protocol):
"""元数据源,按 RSS 订阅的「解析器」(``RSSItem.parser``)选择。"""
async def enrich(self, request: MetadataRequest) -> Metadata | None: ...

View File

@@ -9,6 +9,15 @@ Provider 扩展点(``@provider``):
- ``LLM_PROVIDER``:返回 :class:`ab_sdk.llm.LLMProviderAdapter` 子类
- ``SEARCH_SITE``:返回 :class:`ab_sdk.search.SearchSite`,id 即站点名
- ``SCHEDULED_TASK``:返回 :class:`ab_sdk.tasks.ScheduledTask`
- ``METADATA_PROVIDER``:返回 :class:`ab_sdk.ingest.MetadataProvider`,id 即
RSS 订阅的「解析器」(``RSSItem.parser``)取值
钩子扩展点(``@hook``,契约见 :mod:`ab_sdk.ingest`):
- ``TORRENT_FILTER``(filter):决定已匹配规则的种子是否下载
- ``TITLE_PARSED``(transform):修正标题解析结果
- ``TORRENT_ADDING``(transform):修改发给下载器的添加请求
- ``HTTP_REQUEST``(transform):修改宿主 GET 请求的请求头
"""
DOWNLOADER = "downloader"
@@ -16,3 +25,10 @@ NOTIFIER = "notifier"
LLM_PROVIDER = "llm_provider"
SEARCH_SITE = "search_site"
SCHEDULED_TASK = "scheduled_task"
# --- P3 ingest ---
METADATA_PROVIDER = "metadata_provider"
TORRENT_FILTER = "torrent.filter"
TITLE_PARSED = "title.parsed"
TORRENT_ADDING = "torrent.adding"
HTTP_REQUEST = "http.request"

View File

@@ -114,5 +114,10 @@ async def list_plugin_providers():
"""插件提供的 Provider id(按扩展点),设置页把它们并入下拉候选。"""
return {
point: plugin_provider_ids(point)
for point in (points.DOWNLOADER, points.NOTIFIER, points.SEARCH_SITE)
for point in (
points.DOWNLOADER,
points.NOTIFIER,
points.SEARCH_SITE,
points.METADATA_PROVIDER,
)
}

View File

@@ -20,6 +20,7 @@ from module.parser.analyser.mikan_parser import reset_cache as reset_mikan_cache
from module.parser.analyser.tmdb_parser import reset_cache as reset_tmdb_cache
from module.parser.title_parser import reset_cache as reset_llm_parser
from module.plugin import PluginManager
from module.plugin import host as plugin_host
from module.plugin.host import get_registry
from module.rss import RSSAnalyser
from module.searcher.searcher import reset_cache as reset_poster_cache
@@ -90,6 +91,9 @@ class AppContext:
scheduler, self.plugins.registry, self.plugins.breaker
)
self.plugins.on_change = self._on_plugins_changed
# 模块级代码(RSS 引擎、下载门面、网络层)没有 ctx,经进程级访问器
# 拿到 runner 执行 ingest 钩子
plugin_host.set_runner(self.plugins.runner)
# Downloader-status TTL cache (was ProgramStatus.check_downloader_status).
self._downloader_status = False
self._downloader_reason: str | None = None

View File

@@ -4,10 +4,12 @@ from collections import defaultdict
from urllib.parse import urlparse
from ab_sdk import points
from ab_sdk.ingest import AddRequest
from module.conf import settings
from module.models import Bangumi, Torrent
from module.network import RequestContent
from module.plugin import host as plugin_host
from module.plugin.views import bangumi_info, torrent_info
from .base import (
AddResult,
@@ -50,6 +52,40 @@ async def _fetch_torrent_files(
return [f for group in groups for f in group]
async def _apply_adding_hooks(
torrent: Torrent | list[Torrent],
bangumi: Bangumi,
save_path: str,
category: str,
tags: str | None,
) -> tuple[str, str, str | None]:
"""``torrent.adding``:插件修改保存路径、分类与标签。
``ab:<id>`` 标签是重命名时定位番剧的依据,钩子删掉时补回并告警。
"""
runner = plugin_host.hook_runner(points.TORRENT_ADDING)
if runner is None:
return save_path, category, tags
torrents = torrent if isinstance(torrent, list) else [torrent]
request = AddRequest(
bangumi=bangumi_info(bangumi),
torrents=tuple(torrent_info(t) for t in torrents),
save_path=save_path,
category=category,
tags=tuple(tags.split(",")) if tags else (),
)
result = await runner.transform(points.TORRENT_ADDING, request, expect=AddRequest)
new_tags = [t.strip() for t in result.tags if t.strip()]
if tags and tags not in new_tags:
logger.warning(
"torrent.adding removed required tag %s for %s; restored",
tags,
bangumi.official_title,
)
new_tags.insert(0, tags)
return result.save_path, result.category, ",".join(new_tags) or None
# ---------------------------------------------------------------------------
# Module-level concrete-client cache (session reuse, #1039 / #900)
#
@@ -425,12 +461,17 @@ class DownloadClient:
torrent_url = None
# Create tag with bangumi_id for offset lookup during rename
tags = f"ab:{bangumi.id}" if bangumi.id else None
save_path, category = bangumi.save_path, "Bangumi"
if plugin_host.hook_runner(points.TORRENT_ADDING) is not None:
save_path, category, tags = await _apply_adding_hooks(
torrent, bangumi, save_path, category, tags
)
try:
result = await self.client.add_torrents(
torrent_urls=torrent_url,
torrent_files=torrent_file,
save_path=bangumi.save_path,
category="Bangumi",
save_path=save_path,
category=category,
tags=tags,
)
if result is AddResult.ADDED:

View File

@@ -6,7 +6,10 @@ from typing import Any
import httpx
from httpx_socks import AsyncProxyTransport
from ab_sdk import points
from ab_sdk.ingest import HttpRequest
from module.conf import settings
from module.plugin import host as plugin_host
logger = logging.getLogger(__name__)
@@ -108,6 +111,17 @@ async def reset_shared_client():
_shared_client_proxy_key = None
async def _apply_http_hooks(url: str, headers: dict) -> dict:
"""``http.request``:插件按 URL 修改请求头(私有站 Cookie、UA 等)。"""
runner = plugin_host.hook_runner(points.HTTP_REQUEST)
if runner is None:
return headers
request = HttpRequest(method="GET", url=url, headers=dict(headers))
result = await runner.transform(points.HTTP_REQUEST, request, expect=HttpRequest)
# url / method 不允许被改写,只取请求头
return dict(result.headers)
class RequestURL:
# More complete User-Agent to avoid Cloudflare blocking
DEFAULT_UA = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
@@ -138,7 +152,7 @@ class RequestURL:
self._client is not None
), "RequestURL must be used as an async context manager"
try_time = 0
headers = self._get_headers(url)
headers = await _apply_http_hooks(url, self._get_headers(url))
while True:
try:
req = await self._client.get(url=url, headers=headers)

View File

@@ -2,6 +2,7 @@ import asyncio
import logging
import time
from ab_sdk import points
from module.conf import settings
from module.models import Bangumi, Movie
from module.models.bangumi import Episode
@@ -29,6 +30,7 @@ from module.parser.release_policy import (
normalized_season,
persistence_target,
)
from module.plugin import host as plugin_host
logger = logging.getLogger(__name__)
@@ -327,6 +329,14 @@ def _merge_llm_release(
)
async def _apply_title_hooks(release: ParsedRelease) -> ParsedRelease:
"""``title.parsed``:插件在准入判定前修正解析结果(字幕组别名、季数等)。"""
runner = plugin_host.hook_runner(points.TITLE_PARSED)
if runner is None:
return release
return await runner.transform(points.TITLE_PARSED, release, expect=ParsedRelease)
class TitleParser:
def __init__(self):
pass
@@ -488,6 +498,7 @@ class TitleParser:
if release is None:
return None
release = await _apply_title_hooks(release)
target = persistence_target(release)
if target is None:
logger.debug("Parsed but did not admit resource: %s", raw)

View File

@@ -7,13 +7,20 @@ from typing import Any
from pydantic import BaseModel
from ab_sdk import Event
from module.database import Database
from .bus import EventBus, Handler
PLUGIN_DATA_ROOT = Path("config") / "plugin-data"
def _database():
# 延迟 import:module.plugin.host 被网络层、解析器等底层模块引用,
# 顶层 import module.database 会形成循环依赖
from module.database import Database
return Database()
class DatabaseKV:
"""每次读写开一个独立 session,遵守「session per operation」约定。"""
@@ -21,16 +28,16 @@ class DatabaseKV:
self._plugin_id = plugin_id
async def get(self, key: str, default: Any = None) -> Any:
async with Database() as db:
async with _database() as db:
found, value = await db.plugin_kv.get(self._plugin_id, key)
return value if found else default
async def set(self, key: str, value: Any) -> None:
async with Database() as db:
async with _database() as db:
await db.plugin_kv.set(self._plugin_id, key, value)
async def delete(self, key: str) -> None:
async with Database() as db:
async with _database() as db:
await db.plugin_kv.delete(self._plugin_id, key)

View File

@@ -14,6 +14,7 @@ from ab_sdk import points
from ab_sdk.downloader import DownloaderConnection
from .registry import ExtensionPoint, ExtensionRegistry, ProviderEntry
from .runner import HookRunner
CORE = "core"
@@ -25,9 +26,21 @@ POINTS = (
ExtensionPoint(points.LLM_PROVIDER, "provider", "LLM 解析提供商(llm.provider)"),
ExtensionPoint(points.SEARCH_SITE, "provider", "搜索站点"),
ExtensionPoint(points.SCHEDULED_TASK, "provider", "定时任务"),
# --- P3 ingest ---
ExtensionPoint(
points.METADATA_PROVIDER, "provider", "元数据源(RSS 订阅的 parser)"
),
ExtensionPoint(
points.TORRENT_FILTER, "filter", "已匹配规则的种子是否下载", fail_open=True
),
ExtensionPoint(points.TITLE_PARSED, "transform", "修正标题解析结果"),
ExtensionPoint(points.TORRENT_ADDING, "transform", "修改发给下载器的添加请求"),
ExtensionPoint(points.HTTP_REQUEST, "transform", "修改宿主 GET 请求的请求头"),
)
_registry: ExtensionRegistry | None = None
# 由 AppContext 在构建时设置;未设置时(测试、CLI)钩子一律跳过,只执行宿主逻辑
_runner: HookRunner | None = None
def get_registry() -> ExtensionRegistry:
@@ -47,6 +60,25 @@ def provider(point: str, provider_id: str) -> Any | None:
return entry.factory() if entry is not None else None
def set_runner(runner: HookRunner | None) -> None:
global _runner
_runner = runner
def get_runner() -> HookRunner | None:
return _runner
def hook_runner(point: str) -> HookRunner | None:
"""返回可执行 ``point`` 钩子的 runner;未设置 runner 或该扩展点没有钩子
时返回 None,调用方据此走与无插件时完全相同的路径(逐条种子的热路径
不付出构造快照、调度协程的开销)。"""
runner = _runner
if runner is None or not runner.has_hooks(point):
return None
return runner
def plugin_provider_ids(point: str) -> list[str]:
"""由插件(而非 core)提供的 Provider id,供设置页合并到候选列表。"""
return sorted(
@@ -88,3 +120,18 @@ def _register_core(registry: ExtensionRegistry) -> None:
for provider_id, provider_cls in PROVIDER_REGISTRY.items():
_core(registry, points.NOTIFIER, provider_id, provider_cls)
# 元数据源:沿用 RSSItem.parser 的取值(mikan / tmdb),实现延迟 import
def metadata(name: str):
def factory():
from module.rss import metadata as core_metadata
return core_metadata.CORE_PROVIDERS[name]
return factory
for provider_id in ("mikan", "tmdb"):
registry.add_provider(
points.METADATA_PROVIDER,
ProviderEntry(CORE, provider_id, metadata(provider_id)),
)

View File

@@ -95,6 +95,10 @@ class ExtensionRegistry:
key=lambda e: (rank.get(e.plugin_id, len(rank)), e.priority, e.plugin_id),
)
def has_hooks(self, point_name: str) -> bool:
"""是否有钩子登记在该扩展点上。热路径(逐条种子)用它跳过空扩展点。"""
return bool(self._hooks.get(point_name))
def providers(self, point_name: str) -> dict[str, ProviderEntry]:
self.point(point_name)
return dict(self._providers.get(point_name, {}))

View File

@@ -91,14 +91,49 @@ class HookRunner:
return verdict
return Verdict.ok()
async def transform(self, point: str, value: Any, *args: Any, **kwargs: Any) -> Any:
"""把 ``value`` 依次交给 transform 钩子;钩子失败或返回 None 时沿用原值。"""
def has_hooks(self, point: str) -> bool:
return self._registry.has_hooks(point)
async def transform(
self,
point: str,
value: Any,
*args: Any,
expect: type | tuple[type, ...] | None = None,
**kwargs: Any,
) -> Any:
"""把 ``value`` 依次交给 transform 钩子;钩子失败或返回 None 时沿用原值。
``expect`` 给出时,返回值不是该类型的结果按钩子失败处理(计入熔断),
避免一个插件的错误返回值传给后续钩子和宿主。
"""
for entry in self._registry.hooks(point, self._order(point)):
ok, result = await self._call(entry, point, (value, *args), kwargs)
if ok and result is not None:
value = result
if not ok or result is None:
continue
if expect is not None and not isinstance(result, expect):
reason = f"{point} 返回了 {type(result).__name__},已忽略"
logger.warning("[Plugin:%s] 钩子失败:%s", entry.plugin_id, reason)
self._breaker.record_failure(entry.plugin_id, reason)
continue
value = result
return value
async def call_provider(
self,
plugin_id: str,
point: str,
func: Callable[..., Any],
*args: Any,
timeout: float | None = None,
) -> tuple[bool, Any]:
"""以钩子同样的超时与熔断规则调用插件 Provider 的方法。
返回 ``(是否成功, 结果)``;失败已记录日志并计入熔断。
"""
entry = HookEntry(plugin_id, func, 0, timeout)
return await self._call(entry, point, args, {})
async def _call(
self,
entry: HookEntry,

View File

@@ -0,0 +1,30 @@
"""宿主 ORM 对象 → ``ab_sdk.ingest`` 只读快照的转换。
插件不接触数据库实例,钩子与 Provider 收到的都是这里构造的冻结副本。
"""
from ab_sdk.ingest import BangumiInfo, TorrentInfo
from module.models import Bangumi, Torrent
def torrent_info(torrent: Torrent) -> TorrentInfo:
return TorrentInfo(
name=torrent.name,
url=torrent.url,
homepage=torrent.homepage,
rss_id=torrent.rss_id,
)
def bangumi_info(bangumi: Bangumi) -> BangumiInfo:
return BangumiInfo(
id=bangumi.id,
official_title=bangumi.official_title,
title_raw=bangumi.title_raw,
season=bangumi.season,
group_name=bangumi.group_name,
episode_type=bangumi.episode_type,
rss_link=bangumi.rss_link or "",
save_path=bangumi.save_path,
filter=bangumi.filter or "",
)

View File

@@ -0,0 +1,57 @@
"""内置插件:种子过滤(``torrent.filter``)。
规则自带的「排除过滤」由宿主执行;本插件补充「包含过滤」:配置后,只有
名称匹配任一表达式的种子才会被下载。未配置时不过滤任何种子。
"""
import logging
import re
from pydantic import BaseModel, Field
from ab_sdk import Plugin, Verdict, hook, points
from ab_sdk.ingest import BangumiInfo, Release, TorrentInfo
logger = logging.getLogger(__name__)
class Options(BaseModel):
include: list[str] = Field(
default_factory=list,
title="包含过滤",
description=(
"只下载名称匹配任一表达式的种子(正则,忽略大小写),"
"例如 1080p、简体。留空表示不限制。对所有订阅生效。"
),
)
def compile_terms(terms: list[str]) -> re.Pattern | None:
"""把多个表达式合并为一个模式;非法正则按字面匹配(与排除过滤一致)。"""
cleaned = [t.strip() for t in terms if t.strip()]
if not cleaned:
return None
try:
return re.compile("|".join(cleaned), re.IGNORECASE)
except re.error:
logger.warning("包含过滤 %s 不是合法正则,按字面匹配", cleaned)
return re.compile("|".join(re.escape(t) for t in cleaned), re.IGNORECASE)
class IngestFilters(Plugin[Options]):
config_model = Options
def __init__(self, ctx) -> None:
super().__init__(ctx)
self._include = compile_terms(self.config.include)
@hook(points.TORRENT_FILTER, priority=50)
def include_regex(
self,
torrent: TorrentInfo,
release: Release | None,
bangumi: BangumiInfo,
) -> Verdict:
if self._include is None or self._include.search(torrent.name):
return Verdict.ok()
return Verdict.reject("不匹配包含过滤")

View File

@@ -0,0 +1,8 @@
[plugin]
id = "ingest-filters"
name = "种子过滤"
version = "1.0.0"
sdk = ">=0.3,<1"
entry = "ingest_filters:IngestFilters"
description = "在规则的排除过滤之外,按包含关键词进一步筛选要下载的种子(留空不生效)"
authors = ["AutoBangumi"]

View File

@@ -1,13 +1,13 @@
import logging
import re
from module.conf import settings
from module.models import Bangumi, Movie, ResponseModel, RSSItem, Torrent
from module.network import RequestContent
from module.parser import TitleParser
from module.parser.analyser.selector import parser_engine_snapshot
from .engine import RSSEngine
from .metadata import enrich_bangumi, enrich_movie
logger = logging.getLogger(__name__)
@@ -20,38 +20,9 @@ class RSSAnalyser:
torrent: Torrent,
fetch_poster: bool = True,
):
if not fetch_poster:
pass
elif rss.parser == "mikan":
if not torrent.homepage:
logger.warning("Mikan movie torrent has no homepage info.")
else:
try:
poster_link, official_title = await TitleParser.mikan_parser(
torrent.homepage
)
except AttributeError as e:
logger.warning(
f"Failed to parse Mikan homepage {torrent.homepage}: {e}"
)
else:
movie.poster_link = poster_link
if official_title:
movie.official_title = official_title
elif rss.parser == "tmdb":
tmdb_title, _, year, poster_link = await TitleParser.tmdb_parser(
movie.official_title,
1,
settings.rss_parser.language,
episode_type="movie",
)
movie.official_title = tmdb_title
if year:
try:
movie.year = int(year)
except (ValueError, TypeError):
pass
movie.poster_link = poster_link
# 元数据源由 rss.parser 选择(metadata_provider 扩展点)
if fetch_poster:
await enrich_movie(movie, rss, torrent)
if movie.official_title:
movie.official_title = re.sub(r"[/:.\\]", " ", movie.official_title)
@@ -62,37 +33,8 @@ class RSSAnalyser:
torrent: Torrent,
fetch_poster: bool = True,
):
if not fetch_poster:
pass
elif rss.parser == "mikan":
if not torrent.homepage:
logger.warning("Mikan torrent has no homepage info.")
else:
try:
poster_link, official_title = await TitleParser.mikan_parser(
torrent.homepage
)
except AttributeError as e:
logger.warning(
f"Failed to parse Mikan homepage " f"{torrent.homepage}: {e}"
)
else:
bangumi.poster_link = poster_link
if official_title:
bangumi.official_title = official_title
elif rss.parser == "tmdb":
tmdb_title, season, year, poster_link = await TitleParser.tmdb_parser(
bangumi.official_title,
bangumi.season,
settings.rss_parser.language,
episode_type=bangumi.episode_type,
)
bangumi.official_title = tmdb_title
bangumi.year = year
bangumi.season = season
bangumi.poster_link = poster_link
else:
pass
if fetch_poster:
await enrich_bangumi(bangumi, rss, torrent)
if bangumi.official_title:
bangumi.official_title = re.sub(r"[/:.\\]", " ", bangumi.official_title)

View File

@@ -6,6 +6,7 @@ from datetime import datetime, timezone
from typing import Optional
from urllib.parse import urlparse
from ab_sdk import points
from module.conf import settings
from module.database import Database
from module.database.bangumi import (
@@ -30,6 +31,8 @@ from module.parser.analyser.tokenizer import (
ParsedRelease,
)
from module.parser.release_policy import preference_identity, preference_revision
from module.plugin import host as plugin_host
from module.plugin.views import bangumi_info, torrent_info
logger = logging.getLogger(__name__)
@@ -195,6 +198,47 @@ class RSSEngine:
return matched
return None
@staticmethod
async def plugin_accepts(torrent: Torrent, bangumi: Bangumi) -> bool:
"""``torrent.filter``:在规则自带的排除过滤之后,由插件决定是否下载。
没有插件钩子时直接放行,不做任何额外解析。
"""
runner = plugin_host.hook_runner(points.TORRENT_FILTER)
if runner is None:
return True
try:
release = parse_configured_release_title(torrent.name)
except Exception as e:
# 解析失败不应让过滤阶段中断整轮刷新,钩子收到 None
logger.debug("Cannot parse %s for plugin filters: %s", torrent.name, e)
release = None
verdict = await runner.filter(
points.TORRENT_FILTER,
torrent_info(torrent),
release,
bangumi_info(bangumi),
)
if not verdict.accept:
logger.debug(
"Plugin filter rejected %s: %s", torrent.name, verdict.reason or ""
)
return verdict.accept
async def _apply_plugin_filters(
self, torrents: list[Torrent], matches: list[Optional[Bangumi]]
) -> list[Optional[Bangumi]]:
if plugin_host.hook_runner(points.TORRENT_FILTER) is None:
return matches
result: list[Optional[Bangumi]] = []
for torrent, matched in zip(torrents, matches):
if matched is not None and not await self.plugin_accepts(torrent, matched):
# 与排除过滤一致:被拒的种子不关联番剧(见 match_torrent)
torrent.bangumi_id = None
matched = None
result.append(matched)
return result
@staticmethod
def _select_preference_skips(
matched: list[tuple[Torrent, Bangumi]],
@@ -347,6 +391,7 @@ class RSSEngine:
] = []
for rss_item, (new_torrents, error) in item_results:
matches = [self.match_torrent(t, bangumi_list) for t in new_torrents]
matches = await self._apply_plugin_filters(new_torrents, matches)
item_matches.append((rss_item, new_torrents, error, matches))
skip_ids = self._select_preference_skips(
@@ -426,6 +471,10 @@ class RSSEngine:
bangumi.rss_link, bangumi.filter.replace(",", "|")
)
torrents = [t for t in torrents if release_fits_bangumi(t.name, bangumi)]
if plugin_host.hook_runner(points.TORRENT_FILTER) is not None:
torrents = [
t for t in torrents if await self.plugin_accepts(t, bangumi)
]
if torrents:
async with DownloadClient() as client:
result = await client.add_torrent(torrents, bangumi)

View File

@@ -0,0 +1,149 @@
"""新规则入库前的元数据补全(``metadata_provider`` 扩展点)。
RSS 订阅的 ``parser`` 字段选择元数据源:内置的 ``mikan``(解析 Mikan 番剧
页面的海报与官方标题)与 ``tmdb`` 以 ``core`` 身份登记,插件可登记新的 id。
未登记的取值(如 ``none``)不做任何补全。
"""
import logging
from dataclasses import replace
from ab_sdk import points
from ab_sdk.ingest import Metadata, MetadataProvider, MetadataRequest
from module.conf import settings
from module.models import Bangumi, Movie, RSSItem, Torrent
from module.parser import TitleParser
from module.plugin import host as plugin_host
from module.plugin.views import torrent_info
logger = logging.getLogger(__name__)
class MikanMetadata:
"""从 Mikan 番剧页面读取海报与官方标题。"""
async def enrich(self, request: MetadataRequest) -> Metadata | None:
homepage = request.torrent.homepage
if not homepage:
if request.kind == "movie":
logger.warning("Mikan movie torrent has no homepage info.")
else:
logger.warning("Mikan torrent has no homepage info.")
return None
try:
poster_link, official_title = await TitleParser.mikan_parser(homepage)
except AttributeError as e:
logger.warning(f"Failed to parse Mikan homepage {homepage}: {e}")
return None
return replace(
request.current,
poster_link=poster_link,
official_title=official_title or request.current.official_title,
)
class TmdbMetadata:
"""按标题搜索 TMDB;未命中时沿用原标题,年份与海报置空(与 3.x 一致)。"""
async def enrich(self, request: MetadataRequest) -> Metadata | None:
current = request.current
title, season, year, poster_link = await TitleParser.tmdb_parser(
current.official_title,
current.season,
request.language,
episode_type=request.episode_type,
)
if request.kind == "movie":
season = current.season
return Metadata(
official_title=title, season=season, year=year, poster_link=poster_link
)
CORE_PROVIDERS: dict[str, MetadataProvider] = {
"mikan": MikanMetadata(),
"tmdb": TmdbMetadata(),
}
async def _resolve(parser_id: str, request: MetadataRequest) -> Metadata | None:
entry = (
plugin_host.get_registry().providers(points.METADATA_PROVIDER).get(parser_id)
)
if entry is None:
return None
if entry.plugin_id == plugin_host.CORE:
# 内置源的异常照旧向上抛出,保持 3.x 行为
return await entry.factory().enrich(request)
async def call() -> Metadata | None:
return await entry.factory().enrich(request)
runner = plugin_host.get_runner()
if runner is not None:
ok, result = await runner.call_provider(
entry.plugin_id, points.METADATA_PROVIDER, call
)
if not ok:
return None
else:
try:
result = await call()
except Exception as e:
logger.warning("[Plugin:%s] 元数据源失败:%s", entry.plugin_id, e)
return None
if result is not None and not isinstance(result, Metadata):
logger.warning(
"[Plugin:%s] 元数据源返回了 %s,已忽略",
entry.plugin_id,
type(result).__name__,
)
return None
return result
async def enrich_bangumi(bangumi: Bangumi, rss: RSSItem, torrent: Torrent) -> None:
request = MetadataRequest(
kind="bangumi",
torrent=torrent_info(torrent),
language=settings.rss_parser.language,
episode_type=bangumi.episode_type,
current=Metadata(
official_title=bangumi.official_title,
season=bangumi.season,
year=bangumi.year,
poster_link=bangumi.poster_link,
),
)
result = await _resolve(rss.parser, request)
if result is None:
return
bangumi.official_title = result.official_title
bangumi.season = result.season
bangumi.year = result.year
bangumi.poster_link = result.poster_link
async def enrich_movie(movie: Movie, rss: RSSItem, torrent: Torrent) -> None:
request = MetadataRequest(
kind="movie",
torrent=torrent_info(torrent),
language=settings.rss_parser.language,
episode_type="movie",
current=Metadata(
official_title=movie.official_title,
season=1,
year=str(movie.year) if movie.year is not None else None,
poster_link=movie.poster_link,
),
)
result = await _resolve(rss.parser, request)
if result is None:
return
movie.official_title = result.official_title
if result.year:
try:
movie.year = int(result.year)
except (ValueError, TypeError):
pass
movie.poster_link = result.poster_link

View File

@@ -403,7 +403,12 @@ class TestPluginsApi:
def test_providers(self, authed_client, plugin_ctx, registry):
add_plugin_provider(registry, points.DOWNLOADER, "fake", FakeDownloader)
data = authed_client.get("/api/v1/plugins/providers").json()
assert data == {"downloader": ["fake"], "notifier": [], "search_site": []}
assert data == {
"downloader": ["fake"],
"notifier": [],
"search_site": [],
"metadata_provider": [],
}
def test_config_get_masks_plugin_secrets(self, authed_client, plugin_ctx):
settings.plugins.options["demo"] = {"site": "a", "cookie": "c=1"}

View File

@@ -0,0 +1,601 @@
"""P3 ingest 扩展点:torrent.filter、title.parsed、torrent.adding、
http.request、metadata_provider,以及内置插件 ingest-filters。"""
from dataclasses import replace
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
import pytest_asyncio
from ab_sdk import Verdict, points
from ab_sdk.ingest import (
AddRequest,
BangumiInfo,
HttpRequest,
Metadata,
MetadataRequest,
TorrentInfo,
)
from ab_sdk.testing import create_plugin
from module.conf import settings
from module.database import Database
from module.downloader import AddResult
from module.models import Bangumi, Torrent
from module.models.config import LLM, Plugins
from module.parser import TitleParser
from module.parser.analyser.tokenizer import ParsedRelease
from module.plugin import PluginManager, host
from module.plugin.loader import BUILTIN_ROOT, discover
from module.plugin.registry import HookEntry, ProviderEntry
from module.plugin.runner import CircuitBreaker, HookRunner
from module.rss.analyser import RSSAnalyser
from module.rss.engine import RSSEngine
from test.factories import make_bangumi, make_rss_item, make_torrent
TORRENT_NAME = "[Sub] Mushoku Tensei - 12 [1080p].mkv"
@pytest.fixture
def plugins(monkeypatch):
"""全新的进程级注册表 + 已设置的 runner(模拟 AppContext 构建后的状态)。"""
monkeypatch.setattr(host, "_registry", None)
registry = host.get_registry()
tripped: list[str] = []
breaker = CircuitBreaker(3, on_trip=lambda plugin_id, _: tripped.append(plugin_id))
runner = HookRunner(registry, breaker, default_timeout=1.0)
monkeypatch.setattr(host, "_runner", runner)
return SimpleNamespace(registry=registry, runner=runner, tripped=tripped)
@pytest.fixture
def no_runner(monkeypatch):
monkeypatch.setattr(host, "_registry", None)
monkeypatch.setattr(host, "_runner", None)
return host.get_registry()
def add_hook(registry, point, func, plugin_id="ext", priority=100):
registry.add_hook(point, HookEntry(plugin_id, func, priority, None))
@pytest_asyncio.fixture
async def rss_engine(db_engine):
return RSSEngine(Database(engine=db_engine))
async def refresh_with(engine: RSSEngine, torrents: list[Torrent]):
client = AsyncMock()
client.add_torrent = AsyncMock(return_value=AddResult.ADDED)
with patch.object(RSSEngine, "_get_torrents", new_callable=AsyncMock) as get:
get.return_value = torrents
await engine.refresh_rss(client)
return client
async def seed(engine: RSSEngine, **bangumi_overrides) -> None:
await engine.db.rss.add(make_rss_item(enabled=True))
defaults = {"title_raw": "Mushoku Tensei", "filter": ""}
await engine.db.bangumi.add(make_bangumi(**{**defaults, **bangumi_overrides}))
# ---------------------------------------------------------------- host
class TestHost:
def test_points_declared(self, plugins):
reg = plugins.registry
assert reg.point(points.TORRENT_FILTER).kind == "filter"
assert reg.point(points.TORRENT_FILTER).fail_open is True
for name in (points.TITLE_PARSED, points.TORRENT_ADDING, points.HTTP_REQUEST):
assert reg.point(name).kind == "transform"
providers = reg.providers(points.METADATA_PROVIDER)
assert set(providers) == {"mikan", "tmdb"}
assert all(e.plugin_id == host.CORE for e in providers.values())
def test_hook_runner_fast_path(self, plugins, no_runner):
# no_runner 覆盖了 plugins 设置的 runner
assert host.hook_runner(points.TORRENT_FILTER) is None
def test_hook_runner_requires_hooks(self, plugins):
assert host.hook_runner(points.TORRENT_FILTER) is None
add_hook(plugins.registry, points.TORRENT_FILTER, lambda *a: True)
assert host.hook_runner(points.TORRENT_FILTER) is plugins.runner
async def test_transform_expect_rejects_wrong_type(self, plugins):
add_hook(plugins.registry, points.TITLE_PARSED, lambda r: "oops", "bad")
add_hook(
plugins.registry,
points.TITLE_PARSED,
lambda r: replace(r, group="G"),
"good",
priority=200,
)
release = ParsedRelease(raw="x", title_en="X")
result = await plugins.runner.transform(
points.TITLE_PARSED, release, expect=ParsedRelease
)
assert result == replace(release, group="G")
def test_app_context_sets_runner(self, monkeypatch):
from module.core.context import AppContext
monkeypatch.setattr(host, "_runner", None)
ctx = AppContext.build(settings)
assert host.get_runner() is ctx.plugins.runner
# ---------------------------------------------------------------- torrent.filter
class TestTorrentFilter:
async def test_reject_skips_download_and_unlinks(self, rss_engine, plugins):
await seed(rss_engine)
seen = []
def reject(torrent, release, bangumi):
seen.append((torrent, release, bangumi))
return Verdict.reject("no")
add_hook(plugins.registry, points.TORRENT_FILTER, reject)
client = await refresh_with(
rss_engine, [Torrent(name=TORRENT_NAME, url="https://e/1.torrent")]
)
client.add_torrent.assert_not_called()
stored = await rss_engine.db.torrent.search_all()
assert len(stored) == 1
assert stored[0].bangumi_id is None and not stored[0].downloaded
torrent, release, bangumi = seen[0]
assert isinstance(torrent, TorrentInfo) and torrent.name == TORRENT_NAME
assert isinstance(release, ParsedRelease) and release.episode == 12
assert isinstance(bangumi, BangumiInfo)
assert bangumi.title_raw == "Mushoku Tensei"
async def test_accept_downloads(self, rss_engine, plugins):
await seed(rss_engine)
add_hook(plugins.registry, points.TORRENT_FILTER, lambda *a: True)
client = await refresh_with(
rss_engine, [Torrent(name=TORRENT_NAME, url="https://e/1.torrent")]
)
client.add_torrent.assert_called_once()
async def test_fail_open_on_plugin_error(self, rss_engine, plugins):
await seed(rss_engine)
def boom(*args):
raise RuntimeError("bad plugin")
add_hook(plugins.registry, points.TORRENT_FILTER, boom)
client = await refresh_with(
rss_engine, [Torrent(name=TORRENT_NAME, url="https://e/1.torrent")]
)
client.add_torrent.assert_called_once()
async def test_exclude_filter_runs_before_hooks(self, rss_engine, plugins):
await seed(rss_engine, filter="1080")
hook = MagicMock(return_value=True)
add_hook(plugins.registry, points.TORRENT_FILTER, hook)
client = await refresh_with(
rss_engine, [Torrent(name=TORRENT_NAME, url="https://e/1.torrent")]
)
client.add_torrent.assert_not_called()
hook.assert_not_called()
async def test_no_hook_fast_path_skips_snapshots(self, rss_engine, no_runner):
await seed(rss_engine)
with patch(
"module.rss.engine.torrent_info", side_effect=AssertionError("called")
):
client = await refresh_with(
rss_engine, [Torrent(name=TORRENT_NAME, url="https://e/1.torrent")]
)
client.add_torrent.assert_called_once()
async def test_download_bangumi_applies_filter(self, rss_engine, plugins):
bangumi = make_bangumi(title_raw="Mushoku Tensei", filter="")
add_hook(
plugins.registry,
points.TORRENT_FILTER,
lambda t, r, b: "- 12" not in t.name,
)
torrents = [
Torrent(name=TORRENT_NAME, url="https://e/12.torrent"),
Torrent(
name="[Sub] Mushoku Tensei - 11 [1080p].mkv", url="https://e/11.torrent"
),
]
req = AsyncMock()
req.get_torrents = AsyncMock(return_value=torrents)
client = AsyncMock()
client.add_torrent = AsyncMock(return_value=AddResult.ADDED)
with (
patch("module.rss.engine.RequestContent") as MockReq,
patch("module.rss.engine.DownloadClient") as MockClient,
):
MockReq.return_value.__aenter__ = AsyncMock(return_value=req)
MockReq.return_value.__aexit__ = AsyncMock(return_value=False)
MockClient.return_value.__aenter__ = AsyncMock(return_value=client)
MockClient.return_value.__aexit__ = AsyncMock(return_value=False)
await rss_engine.download_bangumi(bangumi)
sent = client.add_torrent.call_args.args[0]
assert [t.url for t in sent] == ["https://e/11.torrent"]
# ---------------------------------------------------------------- title.parsed
class TestTitleParsed:
RAW = "[Lilith-Raws] Mushoku Tensei - 11 [1080p].mkv"
async def parse(self) -> Bangumi:
with patch.object(settings, "llm", LLM(enable=False)):
result = await TitleParser.raw_parser(self.RAW)
assert isinstance(result, Bangumi)
return result
async def test_hook_rewrites_release(self, plugins):
baseline = await self.parse()
received = []
def fix(release):
received.append(release)
return replace(release, group="Alias", season=2)
add_hook(plugins.registry, points.TITLE_PARSED, fix)
result = await self.parse()
assert isinstance(received[0], ParsedRelease)
assert result.group_name == "Alias"
assert result.season == 2
assert result.title_raw == baseline.title_raw
async def test_bad_hook_keeps_release(self, plugins):
baseline = await self.parse()
def boom(release):
raise ValueError("bad")
add_hook(plugins.registry, points.TITLE_PARSED, boom, "a")
add_hook(plugins.registry, points.TITLE_PARSED, lambda r: {"x": 1}, "b")
result = await self.parse()
assert result.model_dump() == baseline.model_dump()
async def test_hook_can_block_admission(self, plugins):
from module.parser.analyser.tokenizer import MediaType
add_hook(
plugins.registry,
points.TITLE_PARSED,
lambda r: replace(r, media_type=MediaType.PV),
)
with patch.object(settings, "llm", LLM(enable=False)):
assert await TitleParser.raw_parser(self.RAW) is None
# ---------------------------------------------------------------- torrent.adding
@pytest.fixture
def download_client(mock_qb_client):
from module.downloader.download_client import DownloadClient
with patch(
"module.downloader.download_client.DownloadClient._DownloadClient__getClient",
return_value=mock_qb_client,
):
client = DownloadClient()
client.client = mock_qb_client
mock_qb_client.add_torrents.return_value = AddResult.ADDED
return client
async def add_magnet(client, bangumi):
torrent = make_torrent(url="magnet:?xt=urn:btih:abc")
with patch("module.downloader.download_client.RequestContent") as MockReq:
MockReq.return_value.__aenter__ = AsyncMock(return_value=AsyncMock())
MockReq.return_value.__aexit__ = AsyncMock(return_value=False)
return await client.add_torrent(torrent, bangumi)
class TestTorrentAdding:
async def test_no_hook_keeps_defaults(self, download_client, mock_qb_client):
bangumi = make_bangumi(id=7)
assert await add_magnet(download_client, bangumi) is AddResult.ADDED
kwargs = mock_qb_client.add_torrents.call_args.kwargs
assert kwargs["category"] == "Bangumi"
assert kwargs["tags"] == "ab:7"
assert kwargs["save_path"] == bangumi.save_path
async def test_hook_modifies_request_and_tag_restored(
self, plugins, download_client, mock_qb_client
):
received: list[AddRequest] = []
def adjust(request: AddRequest) -> AddRequest:
received.append(request)
return replace(
request,
category="Anime",
save_path=request.save_path + "-x",
tags=("hevc",),
)
add_hook(plugins.registry, points.TORRENT_ADDING, adjust)
bangumi = make_bangumi(id=7)
original_path = bangumi.save_path
assert original_path is not None
await add_magnet(download_client, bangumi)
kwargs = mock_qb_client.add_torrents.call_args.kwargs
assert kwargs["category"] == "Anime"
assert kwargs["save_path"] == original_path + "-x"
assert kwargs["tags"] == "ab:7,hevc"
assert received[0].tags == ("ab:7",)
assert received[0].bangumi.id == 7
assert received[0].torrents[0].url == "magnet:?xt=urn:btih:abc"
# 宿主的 bangumi 对象不被修改
assert bangumi.save_path == original_path
async def test_failing_hook_falls_back(
self, plugins, download_client, mock_qb_client
):
def boom(request):
raise RuntimeError("bad")
add_hook(plugins.registry, points.TORRENT_ADDING, boom)
await add_magnet(download_client, make_bangumi(id=3))
kwargs = mock_qb_client.add_torrents.call_args.kwargs
assert (kwargs["category"], kwargs["tags"]) == ("Bangumi", "ab:3")
# ---------------------------------------------------------------- http.request
class TestHttpRequest:
async def fetch(self, url="https://pt.example/rss"):
from module.network.request_url import RequestURL
req = RequestURL()
response = MagicMock()
req._client = MagicMock()
req._client.get = AsyncMock(return_value=response)
await req.get_url(url)
return req._client.get.call_args.kwargs["headers"]
async def test_no_hook_default_headers(self, no_runner):
headers = await self.fetch()
assert "Cookie" not in headers
assert headers["User-Agent"]
async def test_hook_adds_cookie_per_host(self, plugins):
def cookie(request: HttpRequest):
if "pt.example" not in request.url:
return None
return replace(request, headers={**request.headers, "Cookie": "uid=1"})
add_hook(plugins.registry, points.HTTP_REQUEST, cookie)
assert (await self.fetch())["Cookie"] == "uid=1"
assert "Cookie" not in await self.fetch("https://mikanani.me/rss")
async def test_failing_hook_keeps_headers(self, plugins):
def boom(request):
raise RuntimeError("bad")
add_hook(plugins.registry, points.HTTP_REQUEST, boom)
headers = await self.fetch()
assert "Cookie" not in headers and headers["User-Agent"]
# ---------------------------------------------------------------- metadata
class FakeMetadata:
def __init__(self, result=None, error: Exception | None = None):
self.result = result
self.error = error
self.requests: list[MetadataRequest] = []
async def enrich(self, request: MetadataRequest):
self.requests.append(request)
if self.error is not None:
raise self.error
return self.result
def add_metadata(registry, provider_id, impl, plugin_id="ext"):
registry.add_provider(
points.METADATA_PROVIDER, ProviderEntry(plugin_id, provider_id, lambda: impl)
)
class TestMetadataProvider:
async def test_plugin_provider_selected_by_rss_parser(self, plugins):
impl = FakeMetadata()
impl.result = Metadata(
official_title="葬送的芙莉莲", season=2, year="2023", poster_link="p.jpg"
)
add_metadata(plugins.registry, "bgm", impl)
bangumi = make_bangumi(official_title="Frieren", season=1, year=None)
await RSSAnalyser().official_title_parser(
bangumi, make_rss_item(parser="bgm"), make_torrent()
)
assert (bangumi.official_title, bangumi.season, bangumi.year) == (
"葬送的芙莉莲",
2,
"2023",
)
assert bangumi.poster_link == "p.jpg"
request = impl.requests[0]
assert request.kind == "bangumi"
assert request.current.official_title == "Frieren"
async def test_plugin_provider_failure_keeps_values(self, plugins):
add_metadata(plugins.registry, "bgm", FakeMetadata(error=RuntimeError("x")))
bangumi = make_bangumi(official_title="Frieren")
for _ in range(3):
await RSSAnalyser().official_title_parser(
bangumi, make_rss_item(parser="bgm"), make_torrent()
)
assert bangumi.official_title == "Frieren"
assert plugins.tripped == ["ext"]
async def test_plugin_provider_wrong_type_ignored(self, plugins):
add_metadata(plugins.registry, "bgm", FakeMetadata(result={"title": "x"}))
movie_rss = make_rss_item(parser="bgm")
bangumi = make_bangumi(official_title="Frieren")
await RSSAnalyser().official_title_parser(bangumi, movie_rss, make_torrent())
assert bangumi.official_title == "Frieren"
async def test_unknown_parser_is_noop(self, plugins):
bangumi = make_bangumi(official_title="A.B", poster_link="x")
await RSSAnalyser().official_title_parser(
bangumi, make_rss_item(parser="none"), make_torrent()
)
assert (bangumi.official_title, bangumi.poster_link) == ("A B", "x")
async def test_movie_metadata(self, plugins):
from module.models import Movie
add_metadata(
plugins.registry,
"bgm",
FakeMetadata(Metadata(official_title="Movie", year="2020")),
)
movie = Movie(official_title="Raw", title_raw="Raw", year=None)
await RSSAnalyser().official_title_parser_movie(
movie, make_rss_item(parser="bgm"), make_torrent()
)
assert (movie.official_title, movie.year) == ("Movie", 2020)
async def test_core_tmdb_miss_resets_year(self, plugins):
"""与 3.x 一致:TMDB 未命中时沿用原标题,年份与海报被置空。"""
bangumi = make_bangumi(official_title="Frieren", year="2023", season=1)
with patch.object(
TitleParser,
"tmdb_parser",
AsyncMock(return_value=("Frieren", 1, None, None)),
) as tmdb:
await RSSAnalyser().official_title_parser(
bangumi, make_rss_item(parser="tmdb"), make_torrent()
)
assert tmdb.call_args.args[:2] == ("Frieren", 1)
assert (bangumi.year, bangumi.poster_link) == (None, None)
async def test_core_mikan_keeps_title_when_empty(self, plugins):
bangumi = make_bangumi(official_title="Frieren", year="2023")
with patch.object(
TitleParser, "mikan_parser", AsyncMock(return_value=("poster.jpg", ""))
):
await RSSAnalyser().official_title_parser(
bangumi, make_rss_item(parser="mikan"), make_torrent()
)
assert (bangumi.official_title, bangumi.poster_link, bangumi.year) == (
"Frieren",
"poster.jpg",
"2023",
)
async def test_core_mikan_errors_still_propagate(self, plugins):
bangumi = make_bangumi()
with (
patch.object(
TitleParser, "mikan_parser", AsyncMock(side_effect=KeyError("x"))
),
pytest.raises(KeyError),
):
await RSSAnalyser().official_title_parser(
bangumi, make_rss_item(parser="mikan"), make_torrent()
)
async def test_fetch_poster_false_skips_provider(self, plugins):
impl = FakeMetadata(Metadata(official_title="X"))
add_metadata(plugins.registry, "bgm", impl)
bangumi = make_bangumi(official_title="Frieren")
await RSSAnalyser().official_title_parser(
bangumi, make_rss_item(parser="bgm"), make_torrent(), fetch_poster=False
)
assert impl.requests == []
# ---------------------------------------------------------------- built-in plugin
def load_builtin():
candidates, errors = discover(
local_root=BUILTIN_ROOT / "__missing__", entry_point_group="ab-test-none"
)
assert errors == []
return {c.manifest.id: c for c in candidates}["ingest-filters"]
TORRENT = TorrentInfo(name="[Sub] Show - 01 [1080p][简体].mkv", url="u")
BANGUMI = BangumiInfo(id=1, official_title="Show", title_raw="Show", season=1)
class TestIngestFiltersPlugin:
def plugin_cls(self):
return load_builtin().load()
def test_discovered_as_builtin(self):
candidate = load_builtin()
assert candidate.source == "builtin" and candidate.signed
def test_empty_config_is_noop(self, tmp_path):
plugin, _ = create_plugin(self.plugin_cls(), {}, data_dir=tmp_path)
assert plugin.include_regex(TORRENT, None, BANGUMI).accept
def test_include_regex(self, tmp_path):
cls = self.plugin_cls()
plugin, _ = create_plugin(cls, {"include": ["720p", "简体"]}, data_dir=tmp_path)
assert plugin.include_regex(TORRENT, None, BANGUMI).accept
plugin, _ = create_plugin(cls, {"include": ["2160P "]}, data_dir=tmp_path)
verdict = plugin.include_regex(TORRENT, None, BANGUMI)
assert not verdict.accept and verdict.reason
def test_invalid_regex_falls_back_to_literal(self, tmp_path):
plugin, _ = create_plugin(
self.plugin_cls(), {"include": ["[简体"]}, data_dir=tmp_path
)
assert plugin.include_regex(TORRENT, None, BANGUMI).accept
async def test_loaded_by_manager_enabled_by_default(self, plugins, tmp_path):
settings_obj = SimpleNamespace(
plugins=Plugins(options={"ingest-filters": {"include": ["2160p"]}})
)
manager = PluginManager(
settings_obj,
registry=plugins.registry,
discover_fn=lambda: ([load_builtin()], []),
data_root=tmp_path,
)
await manager.start()
try:
status = {s.id: s for s in manager.statuses()}["ingest-filters"]
assert status.state == "active"
assert status.config_schema is not None
assert "include" in status.config_schema["properties"]
verdict = await manager.runner.filter(
points.TORRENT_FILTER, TORRENT, None, BANGUMI
)
assert not verdict.accept
finally:
await manager.stop()
assert not plugins.registry.has_hooks(points.TORRENT_FILTER)
async def test_default_config_does_not_change_refresh(
self, rss_engine, plugins, tmp_path
):
manager = PluginManager(
SimpleNamespace(plugins=Plugins()),
registry=plugins.registry,
discover_fn=lambda: ([load_builtin()], []),
data_root=tmp_path,
)
await manager.start()
try:
await seed(rss_engine)
client = await refresh_with(
rss_engine, [Torrent(name=TORRENT_NAME, url="https://e/1.torrent")]
)
finally:
await manager.stop()
client.add_torrent.assert_called_once()

View File

@@ -77,3 +77,8 @@ factory
# 动态创建插件包时设置的模块属性
_.__path__
_.__package__
# --- P3 ingest ---
# 内置插件:由插件加载器按 plugin.toml 的 entry 导入,钩子经 @hook 扫描登记
IngestFilters
_.include_regex

View File

@@ -13,8 +13,18 @@
| LLM 解析提供商 | `points.LLM_PROVIDER` | `LLMProviderAdapter` 子类 | 设置 → LLM → 提供商 |
| 搜索站点 | `points.SEARCH_SITE` | `SearchSite` | 搜索框的站点列表 |
| 定时任务 | `points.SCHEDULED_TASK` | `ScheduledTask` | 插件启用即生效 |
| 元数据源 | `points.METADATA_PROVIDER` | `MetadataProvider` | RSS 订阅的「解析器」 |
此外,插件可以订阅事件(`@subscribe`),并使用私有的键值存储和数据目录。RSS 过滤、重命名等流水线扩展点会在后续版本陆续开放。
以及以下钩子(`@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 等) |
此外,插件可以订阅事件(`@subscribe`),并使用私有的键值存储和数据目录。重命名等整理流水线的扩展点会在后续版本开放。
## 目录结构
@@ -31,7 +41,7 @@ config/plugins/local/my-plugin/ # 目录名必须与清单中的 id 相同
id = "my-plugin" # 小写字母、数字、连字符
name = "我的插件"
version = "0.1.0"
sdk = ">=0.2,<1" # 依赖的 ab_sdk 版本范围
sdk = ">=0.3,<1" # 依赖的 ab_sdk 版本范围
entry = "my_plugin:MyPlugin" # 模块:Plugin 子类,相对插件目录
description = "一句话说明"
permissions = ["network"] # 仅用于向用户展示,不做强制
@@ -103,7 +113,7 @@ class MyPlugin(Plugin[Options]):
| 装饰器 | 用途 |
| --- | --- |
| `@provider(point, id=...)` | 工厂方法,返回该扩展点约定的实现 |
| `@hook(point, priority=...)` | 挂到宿主声明的 filter / transform 扩展点(后续版本开放) |
| `@hook(point, priority=...)` | 挂到宿主声明的 filter / transform 扩展点 |
| `@subscribe(kind)` | 订阅事件,`"*"` 表示全部事件 |
扩展点名写错时,插件会加载失败并在插件列表里显示原因,不会被悄悄忽略。
@@ -173,6 +183,121 @@ def sync(self):
任务与 RSS 刷新等内置任务一同启停,失败计入熔断。
### 种子过滤(torrent.filter)
RSS 刷新时,种子先按规则匹配,再经过规则自带的「排除过滤」,最后交给 `torrent.filter` 钩子。任一钩子拒绝,该种子就不下载,也不会关联到番剧(与排除过滤相同)。手动「收集」整季时同样生效。
```python
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 表示不修改。返回其他类型会被忽略并计入熔断。
```python
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`:
```python
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` 的改动会被忽略。
```python
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 按 RSS 订阅的「解析器」取值选择元数据源,补全官方标题、季度、年份和海报。内置 `mikan` 与 `tmdb`,插件可以增加新的取值:
```python
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` 修改。失败或超时时保留原值,并计入熔断。目前 WebUI 的订阅表单只提供内置取值,插件元数据源需要通过 API 设置订阅的 `parser` 字段。
## 测试
`ab_sdk.testing` 可以在不启动 AB 的情况下构造并驱动插件:

View File

@@ -21,6 +21,19 @@
- **扩展点改名**:`search_provider` 改为 `search_site`,以区分用户在搜索设置里维护的站点列表。
- 签名目录来源(LLM 安装器泛化)、`dev_mode` 文件监听、bark / wecom 旧字段别名清理,移到 P7(生态)。
### 实施中的调整(P3)
- **ingest 不拆成独立的 `pipeline/ingest.py`**。钩子与 Provider 直接接入现有 `rss/engine.py`、`parser/title_parser.py`、`downloader/download_client.py`、`network/request_url.py`、`rss/analyser.py`,宿主逻辑原样保留,只在原位置调用扩展点。模块级代码通过 `module.plugin.host.get_runner()` / `hook_runner(point)` 拿到 runner(由 `AppContext` 构建时设置);未设置 runner 或扩展点上没有钩子时直接走原路径,不构造快照。
- **钩子参数是冻结快照,不是 ORM 对象**。`ab_sdk.ingest` 定义 `TorrentInfo`、`BangumiInfo`、`AddRequest`、`HttpRequest`、`Metadata` / `MetadataRequest`;解析结果直接传冻结的 `ParsedRelease`,SDK 只给出只读 Protocol `Release`。transform 钩子返回类型不符时按失败处理并计入熔断(`HookRunner.transform(expect=...)`)。
- **`torrent.filter`** 在规则自带的排除过滤之后执行,排除过滤仍是宿主逻辑,语义不变。被拒的种子与排除过滤一样不关联番剧。手动「收集」整季(`download_bangumi`)同样经过该钩子。
- **`title.parsed`** 只接在 `TitleParser.raw_parser`(新规则创建、收集)的准入判定前。偏好去重 `_select_preference_skips` 使用的同步解析不经过钩子。
- **`torrent.adding`** 允许修改 `save_path`、`category`、`tags`。宿主保证 `ab:<id>` 标签存在(被删时补回)。暂停状态暂不开放,因为 `add_torrents` 契约没有对应参数。
- **`http.request`** 只作用于 `RequestURL.get_url`(RSS、种子文件、站点页面),只能改请求头;POST(通知)不经过。
- **`metadata_provider` 暂不做链式**,仍由 `RSSItem.parser` 选择单个 Provider,列不迁移。内置 mikan / tmdb 以 `core` 登记,行为与 3.x 完全一致,包括 TMDB 未命中时清空年份与海报、mikan 非 `AttributeError` 异常照旧上抛。Provider 返回完整的新 `Metadata`,而不是增量 patch,这样「置空」也能表达。
- **内置插件 `ingest-filters`**(`module/plugins/builtin/ingest-filters/`)只实现全局「包含过滤」,默认启用、留空不生效。种子大小目前不在 RSS 解析结果中(`Torrent` 无 size 列),`size-limit` 推迟到 `feed_source` 解析 enclosure length 之后;按订阅覆盖需要 `Bangumi.plugin_options` 列,同样推迟。
- **推迟**:`feed_source`(与 `FeedConfig.source` 替代 `RSSItem.parser` 一起设计)、`title_parser` 链、`admission_policy`、`bangumi_matcher`、`release_ranker`、`rule.created`、`save_path` Provider,以及第 11 节的批量 `accept_many`。
- `module/plugin/context.py` 改为延迟 import `module.database`,否则网络层引用 `module.plugin.host` 会形成循环依赖。
## 1. 背景与目标
AB 目前只有 **LLM 提供商** 是真正的运行时插件系统:签名下载、目录加载、懒导入、热重载。
@@ -566,7 +579,7 @@ organize: downloader.completed → media_files.classify → file_parser
| **P1 插件运行时 + SDK 骨架** | `ab_sdk` 包(含 `ab_sdk.testing`)、`module/plugin/`(清单、加载、注册表、runner、熔断、EventBus、插件 KV)、`plugins` 配置段、`GET /api/v1/plugins`;SDK 边界测试 | 本地插件可加载、配置、随配置变更重载;已完成。签名目录来源与 LLM 安装器泛化、`dev_mode` 文件监听移到 P2 |
| **P2 迁移已有注册表** | 下载器、通知、LLM、搜索站点、定时任务改为扩展点,内置实现以 `core` 登记;`/api/v1/plugins`(列表、启停、配置、Provider 列表)与 WebUI 插件卡片(JSON Schema 表单);`secret_field` 掩码;插件开发文档 | 已完成;内置行为不变(全量测试)。调整见第 0 节 |
| **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 |
| **P3 流水线插件化:ingest** | `torrent.filter`、`title.parsed`、`torrent.adding`、`http.request` 钩子;`metadata_provider`(mikan / tmdb 以 `core` 登记);内置插件 `ingest-filters`(包含过滤) | 已完成;无插件时行为不变(全量测试)。`feed_source`、`title_parser` 链、`admission_policy`、`matcher`、`ranker`、`save_path`、size 过滤、按订阅覆盖推迟,见第 0 节 |
| **P4 流水线插件化:organize** | `media_files`、`file_parser`、`rename_strategy`(含 `template`)、`conflict_policy`、`file.renamed` 等事件 | 模板重命名;硬链接示例插件 |
| **P5 事件与外部接口** | SSE 改订阅 bus;`api_router`、`mcp_tool` 扩展点;`message_template` | 删掉 SSE 轮询;插件 MCP 工具 |
| **P6 前端插件** | Web Component 挂载点、`AbHost` 桥接、错误边界、`/plugins/<id>/web` 静态资源、`@autobangumi/plugin-ui` 包 | 示例插件「手动选种」以详情页标签形式可用 |

View File

@@ -5,6 +5,7 @@ import type { BangumiRule } from '#/bangumi';
import type { RSS } from '#/rss';
import { rssTemplate } from '#/rss';
import { ruleTemplate } from '#/bangumi';
import { usePluginProviders } from '@/hooks/usePluginProviders';
/** v-model show */
const show = defineModel('show', { default: false });
@@ -17,7 +18,14 @@ const rss = ref<RSS>({ ...rssTemplate });
const rule = defineModel<BangumiRule>('rule', {
default: () => ({ ...ruleTemplate }),
});
const parserTypes = ['tmdb', 'mikan', 'parser'] as const;
const pluginProviders = usePluginProviders();
// 内置元数据源 + 插件登记的元数据源(metadata_provider 扩展点)
const parserTypes = computed(() => [
'tmdb',
'mikan',
'parser',
...pluginProviders.value.metadata_provider,
]);
// UI state
const step = ref<'input' | 'confirm'>('input');

View File

@@ -5,11 +5,12 @@ const providers = ref<PluginProviders>({
downloader: [],
notifier: [],
search_site: [],
metadata_provider: [],
});
let loaded = false;
/**
* 插件提供的下载器 / 通知渠道 / 搜索站点 id,供设置页下拉框合并候选。
* 插件提供的下载器 / 通知渠道 / 搜索站点 / 元数据源 id,供设置页下拉框合并候选。
* 全局只请求一次;请求失败(如未登录)时保持空列表,不影响内置选项。
*/
export function usePluginProviders() {

View File

@@ -57,4 +57,5 @@ export interface PluginProviders {
downloader: string[];
notifier: string[];
search_site: string[];
metadata_provider: string[];
}