mirror of
https://github.com/EstrellaXD/Auto_Bangumi.git
synced 2026-11-01 13:15:59 +08:00
fix(plugins): 处理 Codex 审查的六个问题
- 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 为 "/" 时正确映射完整路径
This commit is contained in:
@@ -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)
|
||||
|
||||
@@ -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))
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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` | 可选:服务器上保存的配置键 |
|
||||
|
||||
## 安装
|
||||
|
||||
@@ -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;填写后忽略配置键",
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -19,7 +19,7 @@
|
||||
|
||||
## 海报
|
||||
|
||||
插件只能拿到宿主给出的公开海报地址,拿不到本地海报文件。因此要发送海报,需要在「设置 → 通知」中配置 `base_url`,并保证 OneBot 实现能访问这个地址;它会自行下载图片。未配置时只发送文字。带图片发送失败时,插件去掉图片只发文字再试一次。
|
||||
插件只能拿到宿主给出的公开海报地址,拿不到本地海报文件。因此要发送海报,需要在「设置 → 通知」中配置 `base_url`,并保证 OneBot 实现能访问这个地址;它会自行下载图片。未配置时只发送文字。海报在文字之后作为单独一条消息发送:发送结果只看文字,图片失败(如 OneBot 实现取不到海报)不影响已送达的文字,也不重发。
|
||||
|
||||
## 安装
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
|
||||
@@ -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:<id>` 标签时为空 |
|
||||
| `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` |
|
||||
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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):
|
||||
|
||||
@@ -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` 只发文字再试一次;令牌失效、限流等其它失败不重试,避免重复发送。
|
||||
|
||||
## 安装
|
||||
|
||||
|
||||
@@ -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"]}
|
||||
)
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -10,7 +10,7 @@
|
||||
|
||||
| 字段 | 示例 | 说明 |
|
||||
| --- | --- | --- |
|
||||
| 媒体根目录 | `/downloads/Bangumi` | 下载器中要去掉的路径前缀(与事件路径同为下载器视角)。不在其下的文件被跳过并记录警告 |
|
||||
| 媒体根目录 | `/downloads/Bangumi` | 下载器中要去掉的路径前缀(与事件路径同为下载器视角)。不在其下的文件被跳过并记录警告;填 `/` 时保留完整路径 |
|
||||
| STRM 目录 | `/downloads/STRM` | 写入 `.strm` 的目录,是 AutoBangumi 看到的路径 |
|
||||
| 地址前缀 | `http://nas:8096/media/` | 拼在相对路径前。含 `://` 时,路径的每一段做百分号编码(空格变为 `%20`);不含时(如本地路径 `/mnt/cloud/`)原样拼接 |
|
||||
|
||||
|
||||
@@ -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_root>.strm,落在 strm_root 之外
|
||||
|
||||
@@ -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"))
|
||||
|
||||
Reference in New Issue
Block a user