From 667f769a2df70bdbf9f69d7c8e35ed9f30dc5e67 Mon Sep 17 00:00:00 2001 From: Estrella Pan Date: Wed, 21 Oct 2026 11:24:07 +0200 Subject: [PATCH] =?UTF-8?q?fix(plugins):=20=E5=A4=84=E7=90=86=20Codex=20?= =?UTF-8?q?=E5=AE=A1=E6=9F=A5=E7=9A=84=E5=85=AD=E4=B8=AA=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - apprise：urls 含令牌，改为 secret_field；secret_field 支持 default_factory，可声明列表 / 字典秘密字段 - run-command：stderr 改为自建管道边读边丢，只保留最后 2000 字节；完成仍只看 proc.wait()，结束后给读取 0.5 秒再关闭管道 - run-command：{files} 按原始路径列表展开，含换行的路径不再拆成两个参数 - 通知：插件渠道 extra 里的已声明字段与内置渠道一样展开 $VAR - slack：只在 HTTP 400 invalid_blocks 时退回纯文字重试，其它失败不重发 - onebot：先发文字，再单独发海报；结果只看文字，去掉重试 - strm-writer：media_root 为 "/" 时正确映射完整路径 --- backend/src/ab_sdk/config.py | 10 +- backend/src/module/notification/resolve.py | 6 +- backend/src/test/test_plugin_extensions.py | 18 ++++ examples/plugins/apprise-notifier/README.md | 2 +- .../apprise_notifier/__init__.py | 5 +- .../tests/test_apprise_notifier.py | 9 +- examples/plugins/onebot-notifier/README.md | 2 +- .../onebot_notifier/__init__.py | 36 +++---- .../tests/test_onebot_notifier.py | 48 +++++---- examples/plugins/run-command/README.md | 4 +- .../run-command/run_command/__init__.py | 101 +++++++++++------- .../run-command/tests/test_run_command.py | 27 +++++ examples/plugins/slack-notifier/README.md | 2 +- .../slack-notifier/slack_notifier/__init__.py | 11 +- .../tests/test_slack_notifier.py | 14 +++ examples/plugins/strm-writer/README.md | 2 +- .../strm-writer/strm_writer/__init__.py | 3 +- .../strm-writer/tests/test_strm_writer.py | 16 ++- 18 files changed, 221 insertions(+), 95 deletions(-) diff --git a/backend/src/ab_sdk/config.py b/backend/src/ab_sdk/config.py index cd765a87..fcac84bc 100644 --- a/backend/src/ab_sdk/config.py +++ b/backend/src/ab_sdk/config.py @@ -5,12 +5,14 @@ from typing import Any from pydantic import Field -def secret_field(default: str = "", *, description: str = "", **kwargs: Any) -> Any: +def secret_field(default: Any = "", *, description: str = "", **kwargs: Any) -> Any: """声明秘密字段（密码、令牌、Cookie 等）。 WebUI 以密码框渲染；读取配置的接口只返回掩码，保存时收到掩码会保留原值。 用法：``cookie: str = secret_field(description="站点 Cookie")``。 + 列表、字典整体声明为秘密时，其中每个值都掩码： + ``urls: list[str] = secret_field(default_factory=list)``。 """ - return Field( - default, description=description, json_schema_extra={"secret": True}, **kwargs - ) + if "default_factory" not in kwargs: + kwargs["default"] = default + return Field(description=description, json_schema_extra={"secret": True}, **kwargs) diff --git a/backend/src/module/notification/resolve.py b/backend/src/module/notification/resolve.py index ff37475d..3f508f5b 100644 --- a/backend/src/module/notification/resolve.py +++ b/backend/src/module/notification/resolve.py @@ -57,13 +57,15 @@ def build_provider(config: "ProviderConfig") -> NotificationProvider | None: # 内置渠道：impl 是 NotificationProvider 子类，直接接收整条配置 return impl(config) # extra = 未知字段 + 已填写的已声明字段（group、icon 等同时是 Bark 的字段， - # 声明后不在 model_extra 里）；type/enabled/template 由宿主处理，不放进去 + # 声明后不在 model_extra 里）；type/enabled/template 由宿主处理，不放进去。 + # 已声明字段按别名取同名属性（token、url 等），与内置渠道一样展开 $VAR unknown = config.model_extra or {} declared = config.model_dump( by_alias=True, exclude_none=True, exclude={"type", "enabled", "template", *unknown}, ) - extra = {k: v for k, v in declared.items() if v != ""} | unknown + expanded = {k: getattr(config, k) for k in declared} + extra = {k: v for k, v in expanded.items() if v != ""} | unknown settings = NotifierSettings(template=config.template, extra=extra) return PluginNotifier(config, impl(settings)) diff --git a/backend/src/test/test_plugin_extensions.py b/backend/src/test/test_plugin_extensions.py index 7413c550..06a1ef30 100644 --- a/backend/src/test/test_plugin_extensions.py +++ b/backend/src/test/test_plugin_extensions.py @@ -187,6 +187,24 @@ class TestNotifier: build_provider(cfg) assert created[0].conf.extra == {"room": "42", "group": "g", "icon": "i"} + def test_build_provider_plugin_extra_expands_env_vars_in_declared_fields( + self, registry, monkeypatch + ): + from module.notification.resolve import build_provider + + created: list[RecordingNotifier] = [] + + def factory(conf: NotifierSettings) -> RecordingNotifier: + created.append(RecordingNotifier(conf)) + return created[-1] + + add_plugin_provider(registry, points.NOTIFIER, "chat", factory) + monkeypatch.setenv("BOT_TOKEN", "abc") + # 内置渠道读 config.token 时展开 $VAR，插件拿到的 extra 也应一致 + cfg = ProviderConfig.model_validate({"type": "chat", "token": "$BOT_TOKEN"}) + build_provider(cfg) + assert created[0].conf.extra == {"token": "abc"} + def test_extra_fields_survive_config_roundtrip(self): cfg = ProviderConfig.model_validate({"type": "chat", "room": "42"}) assert cfg.model_dump(by_alias=True)["room"] == "42" diff --git a/examples/plugins/apprise-notifier/README.md b/examples/plugins/apprise-notifier/README.md index 2077cea2..cf991fad 100644 --- a/examples/plugins/apprise-notifier/README.md +++ b/examples/plugins/apprise-notifier/README.md @@ -23,7 +23,7 @@ docker run -d --name apprise -p 8000:8000 caronc/apprise:latest | 字段 | 说明 | | --- | --- | | `server` | Apprise API 服务器地址，默认 `http://apprise:8000` | -| `urls` | 可选：Apprise URL 列表，如 `tgram://bottoken/ChatID`；填写后忽略配置键 | +| `urls` | 可选：Apprise URL 列表，如 `tgram://bottoken/ChatID`；填写后忽略配置键。URL 含令牌，按秘密字段处理，读取配置时每一项只返回掩码 | | `config_key` | 可选：服务器上保存的配置键 | ## 安装 diff --git a/examples/plugins/apprise-notifier/apprise_notifier/__init__.py b/examples/plugins/apprise-notifier/apprise_notifier/__init__.py index 9b5e6e20..eb0a11f2 100644 --- a/examples/plugins/apprise-notifier/apprise_notifier/__init__.py +++ b/examples/plugins/apprise-notifier/apprise_notifier/__init__.py @@ -14,7 +14,7 @@ from typing import Any import httpx2 from pydantic import BaseModel, Field -from ab_sdk import Plugin, points, provider +from ab_sdk import Plugin, points, provider, secret_field from ab_sdk.notify import NotificationMessage, NotifierFactory, NotifierSettings @@ -22,7 +22,8 @@ class Options(BaseModel): server: str = Field( "http://apprise:8000", title="服务器", description="Apprise API 服务器地址" ) - urls: list[str] = Field( + # URL 里带着令牌，按秘密字段掩码 + urls: list[str] = secret_field( default_factory=list, title="Apprise URL", description="如 tgram://bottoken/ChatID；填写后忽略配置键", diff --git a/examples/plugins/apprise-notifier/tests/test_apprise_notifier.py b/examples/plugins/apprise-notifier/tests/test_apprise_notifier.py index 1d2057d1..a4990246 100644 --- a/examples/plugins/apprise-notifier/tests/test_apprise_notifier.py +++ b/examples/plugins/apprise-notifier/tests/test_apprise_notifier.py @@ -3,7 +3,7 @@ import json import httpx2 import pytest -from apprise_notifier import AppriseNotifier +from apprise_notifier import AppriseNotifier, Options from ab_sdk.notify import NotificationMessage, NotifierSettings from ab_sdk.testing import NotifierContract, create_plugin @@ -61,3 +61,10 @@ def test_missing_urls_and_config_key_fails_without_request(): assert asyncio.run(ch.send(MESSAGE)) is False assert asyncio.run(ch.test()) == (False, "需填写 Apprise URL 或配置键") assert seen == [] + + +def test_urls_is_secret_in_config_schema(): + # URL 里带着令牌：宿主的读配置接口按 schema 中的 secret 标记掩码每一项 + urls = Options.model_json_schema()["properties"]["urls"] + assert urls["secret"] is True + assert urls["type"] == "array" diff --git a/examples/plugins/onebot-notifier/README.md b/examples/plugins/onebot-notifier/README.md index 4f4f26ed..761b8b85 100644 --- a/examples/plugins/onebot-notifier/README.md +++ b/examples/plugins/onebot-notifier/README.md @@ -19,7 +19,7 @@ ## 海报 -插件只能拿到宿主给出的公开海报地址，拿不到本地海报文件。因此要发送海报，需要在「设置 → 通知」中配置 `base_url`，并保证 OneBot 实现能访问这个地址；它会自行下载图片。未配置时只发送文字。带图片发送失败时，插件去掉图片只发文字再试一次。 +插件只能拿到宿主给出的公开海报地址，拿不到本地海报文件。因此要发送海报，需要在「设置 → 通知」中配置 `base_url`，并保证 OneBot 实现能访问这个地址；它会自行下载图片。未配置时只发送文字。海报在文字之后作为单独一条消息发送：发送结果只看文字，图片失败（如 OneBot 实现取不到海报）不影响已送达的文字，也不重发。 ## 安装 diff --git a/examples/plugins/onebot-notifier/onebot_notifier/__init__.py b/examples/plugins/onebot-notifier/onebot_notifier/__init__.py index 8ea94691..257a4834 100644 --- a/examples/plugins/onebot-notifier/onebot_notifier/__init__.py +++ b/examples/plugins/onebot-notifier/onebot_notifier/__init__.py @@ -1,13 +1,14 @@ """OneBot v11 通知渠道。 POST JSON 到 ``{url}/send_private_msg`` 或 ``{url}/send_group_msg``，消息为消息段 -数组：文字，以及（开启海报且宿主给出海报地址时）一张图片。HTTP 2xx 且返回 -``status == "ok"`` 或 ``retcode == 0`` 才算成功。带图片发送失败时（OneBot 实现取不到 -海报会拒绝整条消息），去掉图片只发文字再试一次。 +数组。先发文字；文字成功且开启海报、宿主给出海报地址时，再单独发一条图片消息。 +HTTP 2xx 且返回 ``status == "ok"`` 或 ``retcode == 0`` 才算成功，发送结果只看文字： +OneBot 实现取不到海报时只丢图片，文字已经送达，不重发。 协议文档：https://github.com/botuniverse/onebot-11 """ from collections.abc import Callable +from contextlib import suppress from typing import Any, Literal import httpx2 @@ -46,30 +47,27 @@ class Channel: self.settings = settings self.client_factory = client_factory - def _segments(self, message: NotificationMessage) -> list[dict[str, Any]]: + async def _post(self, message: NotificationMessage) -> tuple[bool, str]: + opts = self.options + if not opts.url or not opts.target: + return False, "未填写 API 地址或目标" # 单集通知的正文已含番剧名；系统事件的标题与正文分开，拼在一起发 text = ( message.body if message.kind == "episode" else f"{message.title}\n{message.body}" ) - segments: list[dict[str, Any]] = [] - # SDK 只给出公开海报 URL（没有本地路径），由 OneBot 实现自行下载 - if self.options.send_poster and message.poster_url: - segments.append({"type": "image", "data": {"file": message.poster_url}}) - segments.append({"type": "text", "data": {"text": text}}) - return segments - - async def _post(self, message: NotificationMessage) -> tuple[bool, str]: - opts = self.options - if not opts.url or not opts.target: - return False, "未填写 API 地址或目标" - segments = self._segments(message) try: async with self.client_factory() as client: - ok, detail = await self._send(client, segments) - if not ok and segments[0]["type"] == "image": - ok, detail = await self._send(client, segments[1:]) + ok, detail = await self._send( + client, [{"type": "text", "data": {"text": text}}] + ) + # SDK 只给出公开海报 URL（没有本地路径），由 OneBot 实现自行下载； + # 图片失败（含请求异常）不影响已送达的文字 + if ok and opts.send_poster and message.poster_url: + image = {"type": "image", "data": {"file": message.poster_url}} + with suppress(httpx2.HTTPError): + await self._send(client, [image]) except httpx2.HTTPError as e: return False, f"请求失败：{e}" return ok, detail diff --git a/examples/plugins/onebot-notifier/tests/test_onebot_notifier.py b/examples/plugins/onebot-notifier/tests/test_onebot_notifier.py index 45aad4c5..01481211 100644 --- a/examples/plugins/onebot-notifier/tests/test_onebot_notifier.py +++ b/examples/plugins/onebot-notifier/tests/test_onebot_notifier.py @@ -68,18 +68,25 @@ def test_send_posts_to_endpoint_with_target(message_type, path, id_key): } +TEXT = [{"type": "text", "data": {"text": "番剧 第1集"}}] +IMAGE = [{"type": "image", "data": {"file": POSTER}}] + + @pytest.mark.parametrize( - ("send_poster", "poster_url", "with_image"), - [(True, POSTER, True), (False, POSTER, False), (True, None, False)], + ("send_poster", "poster_url", "messages"), + [ + (True, POSTER, [TEXT, IMAGE]), + (False, POSTER, [TEXT]), + (True, None, [TEXT]), + ], ) -def test_send_adds_image_segment_only_with_poster(send_poster, poster_url, with_image): +def test_send_poster_goes_as_separate_message_after_text( + send_poster, poster_url, messages +): seen: list[httpx2.Request] = [] ch = channel(200, {"status": "ok"}, {**BASE, "send_poster": send_poster}, seen) assert asyncio.run(ch.send(episode(poster_url))) is True - segments = json.loads(seen[0].content)["message"] - image = {"type": "image", "data": {"file": POSTER}} - assert (segments[0] == image) is with_image - assert segments[-1]["type"] == "text" + assert [json.loads(r.content)["message"] for r in seen] == messages @pytest.mark.parametrize(("token", "header"), [("t0k", "Bearer t0k"), ("", None)]) @@ -103,28 +110,31 @@ def test_send_without_target_returns_false_without_request(): assert seen == [] -# OneBot 实现取不到海报时拒绝整条消息 -def reject_image(body: dict) -> httpx2.Response: - if body["message"][0]["type"] == "image": +# OneBot 实现取不到海报：图片消息返回失败，或请求本身出错 +def image_rejected(body: dict) -> httpx2.Response: + if body["message"] == IMAGE: return httpx2.Response(200, json={"status": "failed", "retcode": 1200}) return httpx2.Response(200, json={"status": "ok", "retcode": 0}) -def test_send_rejected_image_retries_once_with_text_only(): +def image_unreachable(body: dict) -> httpx2.Response: + if body["message"] == IMAGE: + raise httpx2.ReadTimeout("timeout") + return httpx2.Response(200, json={"status": "ok", "retcode": 0}) + + +@pytest.mark.parametrize("respond", [image_rejected, image_unreachable]) +def test_send_image_failure_keeps_text_result_without_retry(respond): seen: list[httpx2.Request] = [] - ch = channel(200, reject_image, seen=seen) + ch = channel(200, respond, seen=seen) assert asyncio.run(ch.send(episode(POSTER))) is True - first, retry = seen - assert json.loads(first.content)["message"][0]["type"] == "image" - assert json.loads(retry.content)["message"] == [ - {"type": "text", "data": {"text": "番剧 第1集"}} - ] + assert [json.loads(r.content)["message"] for r in seen] == [TEXT, IMAGE] -def test_send_text_only_failure_is_not_retried(): +def test_send_text_failure_skips_image_and_is_not_retried(): seen: list[httpx2.Request] = [] ch = channel(200, {"status": "failed", "retcode": 100}, seen=seen) - assert asyncio.run(ch.send(episode())) is False + assert asyncio.run(ch.send(episode(POSTER))) is False assert len(seen) == 1 diff --git a/examples/plugins/run-command/README.md b/examples/plugins/run-command/README.md index cac3858e..cee59ee9 100644 --- a/examples/plugins/run-command/README.md +++ b/examples/plugins/run-command/README.md @@ -4,7 +4,7 @@ - `@subscribe("*", timeout=...)` 订阅全部事件，按配置过滤；处理超时放宽到命令超时之上 - `asyncio.create_subprocess_exec` 直接执行参数列表，不经过 shell -- 命令超时被杀掉，退出码非 0 时记录 stderr 末尾；失败只记日志，不让插件被熔断 +- 命令超时被杀掉，退出码非 0 时记录 stderr 的最后 2000 字节（边读边丢，输出再多也不占满内存）；命令退出后插件即关闭 stderr，留在后台的进程需自行重定向输出。失败只记日志，不让插件被熔断 ## 配置 @@ -32,7 +32,7 @@ ln -f "$AB_NEW_PATH" "/media/links/$(basename "$AB_NEW_PATH")" | `AB_BANGUMI_ID` | 全部 | 番剧 id；旧种子没有 `ab:` 标签时为空 | | `AB_DOWNLOADER_ID` | 全部 | 下载器实例 id | | `AB_TORRENT_HASH` | `torrent.organized` | 种子 hash | -| `AB_FILES` | `torrent.organized` | 正片与字幕的最终路径，每行一个 | +| `AB_FILES` | `torrent.organized` | 正片与字幕的最终路径，每行一个。路径本身含换行时无法按行区分，需要准确的路径请用 `{files}` 参数 | | `AB_OLD_PATH` / `AB_NEW_PATH` | `file.renamed` | 重命名前后的路径 | | `AB_FILE_KIND` | `file.renamed` | `media` 或 `subtitle` | diff --git a/examples/plugins/run-command/run_command/__init__.py b/examples/plugins/run-command/run_command/__init__.py index a52388b7..4dbc70dc 100644 --- a/examples/plugins/run-command/run_command/__init__.py +++ b/examples/plugins/run-command/run_command/__init__.py @@ -12,14 +12,13 @@ - ``torrent.organized``：``AB_TORRENT_HASH``、``AB_FILES``（正片与字幕的路径，每行一个） - ``file.renamed``：``AB_OLD_PATH``、``AB_NEW_PATH``、``AB_FILE_KIND`` -路径是下载器视角。命令超时被杀掉；退出码非 0 时记录 stderr 末尾。命令的失败只记日志， -不抛给宿主：外部脚本出错不应让插件被熔断。 +路径是下载器视角。命令超时被杀掉；退出码非 0 时记录 stderr 的最后 2000 字节。 +命令的失败只记日志，不抛给宿主：外部脚本出错不应让插件被熔断。 """ import asyncio import os import re -import tempfile from subprocess import DEVNULL from pydantic import BaseModel, Field @@ -28,6 +27,7 @@ from ab_sdk import Event, Plugin, subscribe from ab_sdk.events import FileRenamed, TorrentOrganized MAX_TIMEOUT = 3600 +STDERR_TAIL = 2000 _PLACEHOLDER = re.compile(r"\{(\w+)\}") @@ -75,10 +75,13 @@ def event_fields(event: Event) -> dict[str, str] | None: } -def build_argv(command: list[str], fields: dict[str, str]) -> list[str]: +def build_argv( + command: list[str], fields: dict[str, str], files: list[str] +) -> list[str]: """整项为 ``{字段}`` 的参数替换为字段值，``{files}`` 展开为每个文件一个参数。 - 部分包含占位符的参数与未知的占位符原样保留。 + ``{files}`` 取原始路径列表，而不是按行拆 ``AB_FILES``：路径里的换行不会拆出 + 多余的参数。部分包含占位符的参数与未知的占位符原样保留。 """ argv: list[str] = [] for arg in command: @@ -86,12 +89,19 @@ def build_argv(command: list[str], fields: dict[str, str]) -> list[str]: if m is None or m[1] not in fields: argv.append(arg) elif m[1] == "files": - argv.extend(fields["files"].splitlines()) + argv.extend(files) else: argv.append(fields[m[1]]) return argv +async def drain_tail(stream: asyncio.StreamReader, tail: bytearray) -> None: + """持续读完 ``stream``，只保留最后 ``STDERR_TAIL`` 字节，内存有上限。""" + while chunk := await stream.read(65536): + tail += chunk + del tail[:-STDERR_TAIL] + + def build_env(fields: dict[str, str]) -> dict[str, str]: """继承宿主环境（否则子进程没有 PATH），再加上 ``AB_*``。 @@ -112,38 +122,57 @@ class RunCommand(Plugin[Options]): fields = event_fields(event) if fields is None: return - argv = build_argv(cfg.command, fields) - # stderr 写进临时文件而不是管道：命令留在后台的子进程继承 stderr 时， - # 读管道要等到它们退出，命令本身早已结束也会被当成超时 - with tempfile.TemporaryFile() as err: - try: - proc = await asyncio.create_subprocess_exec( - *argv, - stdin=DEVNULL, - stdout=DEVNULL, - stderr=err, - env=build_env(fields), - ) - except OSError as e: - self.ctx.log.warning("无法启动命令 %s：%s", argv[0], e) - return - try: - async with asyncio.timeout(cfg.timeout): - await proc.wait() - except TimeoutError: - self.ctx.log.warning("命令 %s 超过 %s 秒，已终止", argv[0], cfg.timeout) - return - finally: - # 超时或宿主取消处理时，不留下孤儿进程 - # ponytail: 只杀直接子进程；命令自己再派生的进程需由脚本负责 - if proc.returncode is None: - proc.kill() - await proc.wait() - err.seek(max(0, err.seek(0, os.SEEK_END) - 2000)) - tail = err.read().decode(errors="replace").strip() + files = ( + [f.path for f in event.files] if isinstance(event, TorrentOrganized) else [] + ) + argv = build_argv(cfg.command, fields, files) + # 自建管道而不是 stderr=PIPE：读端由插件自己关闭。命令留在后台的子进程继承了 + # stderr 时，管道要等它们退出才会关闭，不能让宿主一直挂着它 + stream = asyncio.StreamReader() + r, w = os.pipe() + pipe, _ = await asyncio.get_running_loop().connect_read_pipe( + lambda: asyncio.StreamReaderProtocol(stream), os.fdopen(r, "rb") + ) + try: + proc = await asyncio.create_subprocess_exec( + *argv, + stdin=DEVNULL, + stdout=DEVNULL, + stderr=w, + env=build_env(fields), + ) + except OSError as e: + pipe.close() + self.ctx.log.warning("无法启动命令 %s：%s", argv[0], e) + return + finally: + os.close(w) + # 边读边丢：命令狂写 stderr 时既不会写满管道卡住，也不会占满内存或磁盘 + tail = bytearray() + reader = asyncio.create_task(drain_tail(stream, tail)) + try: + # 是否完成只看 proc.wait()，不等 stderr 读到 EOF + async with asyncio.timeout(cfg.timeout): + await proc.wait() + except TimeoutError: + self.ctx.log.warning("命令 %s 超过 %s 秒，已终止", argv[0], cfg.timeout) + return + finally: + # 超时或宿主取消处理时，不留下孤儿进程 + # ponytail: 只杀直接子进程；命令自己再派生的进程需由脚本负责 + if proc.returncode is None: + proc.kill() + await proc.wait() + # 给读取一点时间读完管道里剩下的内容，后台子进程不关管道时放弃 + await asyncio.wait({reader}, timeout=0.5) + reader.cancel() + pipe.close() if proc.returncode: self.ctx.log.warning( - "命令 %s 退出码 %s：%s", argv[0], proc.returncode, tail + "命令 %s 退出码 %s：%s", + argv[0], + proc.returncode, + tail.decode(errors="replace").strip(), ) else: self.ctx.log.info("命令 %s 已完成（%s）", argv[0], event.kind) diff --git a/examples/plugins/run-command/tests/test_run_command.py b/examples/plugins/run-command/tests/test_run_command.py index 39501d70..1f83a128 100644 --- a/examples/plugins/run-command/tests/test_run_command.py +++ b/examples/plugins/run-command/tests/test_run_command.py @@ -110,6 +110,33 @@ def test_on_event_background_child_does_not_delay_success(caplog): assert "已完成" in caplog.text and "已终止" not in caplog.text +def test_on_event_files_placeholder_keeps_newline_in_path(tmp_path): + # {files} 取原始路径列表：含换行的路径仍是一个参数（AB_FILES 只能按行拼接） + out = tmp_path / "out.json" + event = TorrentOrganized( + torrent_hash="abc", + bangumi_id=1, + files=(OrganizedFile("/dl/a\nb.mkv", "media"),), + ) + run({"command": [sys.executable, "-c", DUMP, str(out), "{files}"]}, event) + assert json.loads(out.read_text())["argv"] == ["/dl/a\nb.mkv"] + + +def test_on_event_huge_stderr_keeps_only_tail(caplog): + # 输出远超管道缓冲：边读边丢，命令不会卡到超时，日志只带最后 2000 字节 + script = ( + "import sys; sys.stderr.write('x' * 5_000_000 + 'END'); sys.stderr.flush(); " + "sys.exit(3)" + ) + start = time.monotonic() + with caplog.at_level(logging.WARNING): + run({"command": [sys.executable, "-c", script], "timeout": 10}) + assert time.monotonic() - start < 5 + [record] = caplog.records + assert "退出码 3" in record.getMessage() + assert record.args[-1] == "x" * 1997 + "END" + + def test_on_event_nonzero_exit_logs_code_and_stderr(caplog): script = "import sys; sys.stderr.write('boom'); sys.exit(3)" with caplog.at_level(logging.WARNING): diff --git a/examples/plugins/slack-notifier/README.md b/examples/plugins/slack-notifier/README.md index 2f5b1ce5..f6532b5d 100644 --- a/examples/plugins/slack-notifier/README.md +++ b/examples/plugins/slack-notifier/README.md @@ -9,7 +9,7 @@ ## 消息格式 -请求体为 `{"text": "*标题*\n正文"}`（Slack 的 mrkdwn，`&`、`<`、`>` 已转义）。配置了 `notification.base_url` 时单集通知带海报：再加 `blocks`（一段文字和一张图片），`text` 作为手机通知的摘要。海报地址必须能被 Slack 访问；Slack 取不到图片时会拒绝整条消息，插件会去掉 `blocks` 只发文字再试一次。 +请求体为 `{"text": "*标题*\n正文"}`（Slack 的 mrkdwn，`&`、`<`、`>` 已转义）。配置了 `notification.base_url` 时单集通知带海报：再加 `blocks`（一段文字和一张图片），`text` 作为手机通知的摘要。海报地址必须能被 Slack 访问；Slack 取不到图片时会以 HTTP 400 `invalid_blocks` 拒绝整条消息，插件只在这种情况下去掉 `blocks` 只发文字再试一次；令牌失效、限流等其它失败不重试，避免重复发送。 ## 安装 diff --git a/examples/plugins/slack-notifier/slack_notifier/__init__.py b/examples/plugins/slack-notifier/slack_notifier/__init__.py index 27c94d4e..02c04809 100644 --- a/examples/plugins/slack-notifier/slack_notifier/__init__.py +++ b/examples/plugins/slack-notifier/slack_notifier/__init__.py @@ -3,8 +3,8 @@ 向 Incoming Webhook POST ``{"text": "*标题*\\n正文"}``（Slack 的 mrkdwn）。带海报时 再加 ``blocks``：一段文字加一张图片，``text`` 留作手机通知里的摘要。Slack 成功时 返回 HTTP 200 与响应体 ``ok``，失败时返回 4xx 与错误码（如 ``invalid_token``）。 -Slack 取不到海报图片时会拒绝整条消息（400 ``invalid_blocks``），此时去掉 ``blocks`` -只发文字再试一次。 +Slack 取不到海报图片时会拒绝整条消息（HTTP 400 ``invalid_blocks``），只有这种情况 +才去掉 ``blocks`` 只发文字再试一次；令牌失效、限流等其它失败不重试，以免重复发送。 """ from collections.abc import Callable @@ -48,6 +48,11 @@ def _ok(resp: httpx2.Response) -> bool: return resp.is_success and resp.text == "ok" +def _blocks_rejected(resp: httpx2.Response) -> bool: + """Slack 因 blocks（如下载不到图片）拒绝消息；``invalid_blocks_format`` 同属此类。""" + return resp.status_code == 400 and "invalid_blocks" in resp.text + + class Channel: def __init__( self, @@ -66,7 +71,7 @@ class Channel: try: async with self.client_factory() as client: resp = await client.post(self.options.webhook_url, json=body) - if not _ok(resp) and "blocks" in body: + if "blocks" in body and _blocks_rejected(resp): resp = await client.post( self.options.webhook_url, json={"text": body["text"]} ) diff --git a/examples/plugins/slack-notifier/tests/test_slack_notifier.py b/examples/plugins/slack-notifier/tests/test_slack_notifier.py index 8d683a4d..ee3bdc13 100644 --- a/examples/plugins/slack-notifier/tests/test_slack_notifier.py +++ b/examples/plugins/slack-notifier/tests/test_slack_notifier.py @@ -93,3 +93,17 @@ def test_send_text_only_failure_is_not_retried(): ch = channel(httpx2.Response(400, text="invalid_payload"), seen) assert asyncio.run(ch.send(message)) is False assert len(seen) == 1 + + +@pytest.mark.parametrize( + ("status", "text"), [(403, "invalid_token"), (429, "rate_limited")] +) +def test_send_with_poster_other_failure_is_not_retried(status, text): + # 只有 blocks 被拒才退回纯文字；其它失败重发可能重复投递 + seen: list[httpx2.Request] = [] + message = NotificationMessage( + kind="episode", title="T", body="b", poster_url="https://p/1.jpg" + ) + ch = channel(httpx2.Response(status, text=text), seen) + assert asyncio.run(ch.send(message)) is False + assert len(seen) == 1 diff --git a/examples/plugins/strm-writer/README.md b/examples/plugins/strm-writer/README.md index e89cf363..fb4ade60 100644 --- a/examples/plugins/strm-writer/README.md +++ b/examples/plugins/strm-writer/README.md @@ -10,7 +10,7 @@ | 字段 | 示例 | 说明 | | --- | --- | --- | -| 媒体根目录 | `/downloads/Bangumi` | 下载器中要去掉的路径前缀（与事件路径同为下载器视角）。不在其下的文件被跳过并记录警告 | +| 媒体根目录 | `/downloads/Bangumi` | 下载器中要去掉的路径前缀（与事件路径同为下载器视角）。不在其下的文件被跳过并记录警告；填 `/` 时保留完整路径 | | STRM 目录 | `/downloads/STRM` | 写入 `.strm` 的目录，是 AutoBangumi 看到的路径 | | 地址前缀 | `http://nas:8096/media/` | 拼在相对路径前。含 `://` 时，路径的每一段做百分号编码（空格变为 `%20`）；不含时（如本地路径 `/mnt/cloud/`）原样拼接 | diff --git a/examples/plugins/strm-writer/strm_writer/__init__.py b/examples/plugins/strm-writer/strm_writer/__init__.py index 4f9c9720..d12f8a78 100644 --- a/examples/plugins/strm-writer/strm_writer/__init__.py +++ b/examples/plugins/strm-writer/strm_writer/__init__.py @@ -40,8 +40,9 @@ class Options(BaseModel): def relative(path: str, media_root: str) -> str | None: """``path`` 不在 ``media_root`` 下（或就是 ``media_root`` 本身）时返回 None。""" + # media_root 为 "/" 时 root 为空串，前缀 "/" 照样匹配所有绝对路径 root = media_root.rstrip("/") - if not root or not path.startswith(root + "/"): + if not media_root or not path.startswith(root + "/"): return None rel = posixpath.normpath(path[len(root) + 1 :]) # "." 会写出 .strm，落在 strm_root 之外 diff --git a/examples/plugins/strm-writer/tests/test_strm_writer.py b/examples/plugins/strm-writer/tests/test_strm_writer.py index 8cfbfd4d..0fbcb607 100644 --- a/examples/plugins/strm-writer/tests/test_strm_writer.py +++ b/examples/plugins/strm-writer/tests/test_strm_writer.py @@ -14,11 +14,16 @@ def organized(*files: OrganizedFile) -> TorrentOrganized: return TorrentOrganized(torrent_hash="h", bangumi_id=1, files=files) -def run(tmp_path, url_prefix: str, *files: OrganizedFile) -> None: +def run( + tmp_path, + url_prefix: str, + *files: OrganizedFile, + media_root: str = "/downloads/Bangumi/", +) -> None: options = { "strm_root": str(tmp_path / "STRM"), "url_prefix": url_prefix, - "media_root": "/downloads/Bangumi/", + "media_root": media_root, } plugin, _ = create_plugin(StrmWriter, options) asyncio.run(plugin.on_organized(organized(*files))) @@ -46,6 +51,13 @@ def test_on_organized_mirrors_media_path_into_strm(tmp_path, url_prefix, content assert (season / "A S01E01.strm").read_text(encoding="utf-8") == content +def test_on_organized_media_root_slash_maps_whole_path(tmp_path): + # 根目录 "/" 去掉尾部斜杠后是空串，曾把所有绝对路径都判为不在根目录下 + run(tmp_path, "/mnt/", OrganizedFile("/dl/A/B.mkv", "media"), media_root="/") + strm = tmp_path / "STRM" / "dl" / "A" / "B.strm" + assert strm.read_text(encoding="utf-8") == "/mnt/dl/A/B.mkv" + + def test_on_organized_prefix_change_rewrites_strm(tmp_path): run(tmp_path, "http://old/", OrganizedFile(MEDIA, "media")) run(tmp_path, "http://new/", OrganizedFile(MEDIA, "media"))