sansenhoshiandClaude Code
4badcfcf32
- policy.py：per-group 正交策略（自动解析 / 自动策略 / 禁用策略 / 存储 A·B·C /
公网 / 链接 / 群文件 + 平台限定），list.json v1/v2 → v3 自动迁移，
写入统一走 PolicyStore（加锁 + .tmp 原子替换 + 字段归一）
- 群文件并行通道 group_file.py：打包 zip（可选 pyzipper AES-256）后优先走 S3 预签名、
本地直传兜底；设了密码但 pyzipper 不可用就放弃上传，不退化成明文
- list_proc.py 收敛到「视频策略」统一入口，权限判定改走 policy
- Web 管理页 /hub/video_analysis（群策略 + 链接解析面板）与 services/web_jobs.py
（只复用纯函数层，Web 上下文不发消息；内存任务表 + 并发闸门 + 超时）
- 媒体命名统一到 utils.py（{作者}_{作者id}/{作品名}[_短码]），cleanup 回收空目录
- 测试：policy / 命名 / 群文件 / web_jobs 四组
顺带 pyproject 的 pytest 加 testpaths=tests（避免收进 debug/ 下的调试脚本）。
Co-Authored-By: Claude Code <noreply@anthropic.com>
304 lines
11 KiB
Python
304 lines
11 KiB
Python
"""群文件打包/投递 (group_file.py) 单元测试
|
|
|
|
group_file.py 只依赖标准库 + nonebot.logger，故用 importlib 按文件路径裸加载
|
|
（与 test_video_policy.py 同法）。build_archive 一律显式传 out_dir —— 默认目录
|
|
要经相对导入取 utils.get_temp_root()，裸加载模块没有包上下文。
|
|
"""
|
|
|
|
import importlib.util
|
|
import sys
|
|
import zipfile
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
_MODULE_PATH = (
|
|
Path(__file__).resolve().parents[1]
|
|
/ "hexi"
|
|
/ "plugins"
|
|
/ "nonebot_plugin_video_analysis"
|
|
/ "services"
|
|
/ "storage"
|
|
/ "group_file.py"
|
|
)
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def gf():
|
|
"""以独立模块名加载 group_file.py，避免与插件包 __init__ 冲突"""
|
|
spec = importlib.util.spec_from_file_location(
|
|
"video_group_file_under_test", _MODULE_PATH
|
|
)
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules[spec.name] = module
|
|
spec.loader.exec_module(module)
|
|
yield module
|
|
sys.modules.pop(spec.name, None)
|
|
|
|
|
|
def _make_files(directory: Path, names: list[str]) -> list[Path]:
|
|
directory.mkdir(parents=True, exist_ok=True)
|
|
made = []
|
|
for name in names:
|
|
path = directory / name
|
|
path.write_bytes(f"hello {name}".encode())
|
|
made.append(path)
|
|
return made
|
|
|
|
|
|
def test_plain_archive(gf, tmp_path):
|
|
files = _make_files(tmp_path / "src", ["a.txt", "b.txt"])
|
|
archive = gf.build_archive(files, "我的作品", out_dir=tmp_path / "out")
|
|
|
|
assert archive.exists() and archive.suffix == ".zip"
|
|
assert "我的作品" in archive.name
|
|
with zipfile.ZipFile(archive) as zf:
|
|
assert sorted(zf.namelist()) == ["a.txt", "b.txt"]
|
|
assert zf.read("a.txt") == b"hello a.txt"
|
|
|
|
|
|
def test_encrypted_archive_requires_password(gf, tmp_path):
|
|
pyzipper = pytest.importorskip("pyzipper")
|
|
files = _make_files(tmp_path / "src", ["v.mp4"])
|
|
archive = gf.build_archive(
|
|
files, "加密作品", password="pw123", out_dir=tmp_path / "out"
|
|
)
|
|
|
|
# 标准库打不开加密包
|
|
with zipfile.ZipFile(archive) as zf:
|
|
with pytest.raises(RuntimeError):
|
|
zf.read("v.mp4")
|
|
|
|
# 正确密码可读，错误密码不行
|
|
with pyzipper.AESZipFile(archive) as zf:
|
|
zf.setpassword(b"pw123")
|
|
assert zf.read("v.mp4") == b"hello v.mp4"
|
|
with pyzipper.AESZipFile(archive) as zf:
|
|
zf.setpassword(b"wrong")
|
|
with pytest.raises(RuntimeError):
|
|
zf.read("v.mp4")
|
|
|
|
|
|
def test_duplicate_names_indexed(gf, tmp_path):
|
|
first = _make_files(tmp_path / "p1", ["same.txt"])[0]
|
|
second = _make_files(tmp_path / "p2", ["same.txt"])[0]
|
|
archive = gf.build_archive([first, second], "t", out_dir=tmp_path / "out")
|
|
|
|
with zipfile.ZipFile(archive) as zf:
|
|
assert sorted(zf.namelist()) == ["same.txt", "same_2.txt"]
|
|
assert zf.read("same_2.txt") == b"hello same.txt"
|
|
|
|
|
|
def test_output_path_is_unique(gf, tmp_path):
|
|
files = _make_files(tmp_path / "src", ["a.txt"])
|
|
out = tmp_path / "out"
|
|
one = gf.build_archive(files, "t", out_dir=out)
|
|
two = gf.build_archive(files, "t", out_dir=out)
|
|
assert one != two and one.exists() and two.exists()
|
|
|
|
|
|
def test_placeholder_title_falls_back(gf, tmp_path):
|
|
"""universal.py 传的 title 是占位符 "title"，不该出现在文件名里"""
|
|
files = _make_files(tmp_path / "src", ["a.txt"])
|
|
archive = gf.build_archive(files, "title", out_dir=tmp_path / "out")
|
|
assert archive.name.startswith("群文件_")
|
|
assert "群文件_群文件" not in archive.name
|
|
|
|
|
|
def test_title_with_path_separators_is_sanitized(gf, tmp_path):
|
|
"""原始标题（抖音文案/YouTube 标题）可能带 /、换行、#话题，不能进文件名"""
|
|
files = _make_files(tmp_path / "src", ["a.txt"])
|
|
archive = gf.build_archive(files, "和/或 #话题\n测试", out_dir=tmp_path / "out")
|
|
|
|
assert "/" not in archive.name and "\\" not in archive.name
|
|
assert "\n" not in archive.name and "#" not in archive.name
|
|
assert archive.name.startswith("和_或_测试")
|
|
|
|
|
|
def test_title_starting_with_archive_prefix_falls_back(gf, tmp_path):
|
|
"""标题本身以「群文件」开头时不要产出 群文件_xxx_群文件_yyy.zip"""
|
|
files = _make_files(tmp_path / "src", ["a.txt"])
|
|
archive = gf.build_archive(files, "群文件_120606", out_dir=tmp_path / "out")
|
|
assert archive.name.startswith("群文件_")
|
|
assert "群文件_120606_群文件" not in archive.name
|
|
|
|
|
|
def test_empty_or_missing_input(gf, tmp_path):
|
|
with pytest.raises(ValueError):
|
|
gf.build_archive([], "t", out_dir=tmp_path / "out")
|
|
with pytest.raises(FileNotFoundError):
|
|
gf.build_archive([tmp_path / "nope.mp4"], "t", out_dir=tmp_path / "out")
|
|
|
|
|
|
def test_password_without_pyzipper_fails_loud(gf, tmp_path, monkeypatch):
|
|
"""设了密码但加密库不可用 → 报错（调用方会放弃上传，绝不能退化传明文）"""
|
|
monkeypatch.setattr(gf, "pyzipper", None)
|
|
assert gf.encryption_available() is False
|
|
|
|
files = _make_files(tmp_path / "src", ["a.txt"])
|
|
with pytest.raises(RuntimeError):
|
|
gf.build_archive(files, "t", password="pw", out_dir=tmp_path / "out")
|
|
# 不加密那一路不受影响
|
|
assert gf.build_archive(files, "t", out_dir=tmp_path / "out").exists()
|
|
|
|
|
|
# ───────────────────── 上传：URI 直传 + S3 降级 ─────────────────────
|
|
|
|
|
|
def test_file_uri_keeps_unicode_and_uses_slashes(gf, tmp_path):
|
|
target = tmp_path / "dir with space" / "群文件_120606.zip"
|
|
target.parent.mkdir(parents=True)
|
|
target.write_bytes(b"x")
|
|
|
|
uri = gf._file_uri(target)
|
|
assert uri.startswith("file:///")
|
|
assert "\\" not in uri
|
|
# 中文/空格保持原样（不做百分号转义，NapCat 侧直接当路径解析）
|
|
assert uri.endswith("/dir with space/群文件_120606.zip")
|
|
|
|
|
|
async def test_s3_link_is_the_primary_channel(gf, tmp_path, monkeypatch):
|
|
"""默认通道就是 S3 链接（本环境本地直传必挂，不先浪费一次调用）"""
|
|
calls: list = []
|
|
link = "http://192.168.2.15:5246/PLANA/x.zip?sig=1"
|
|
|
|
async def fake_call(file_value, group_id, name, folder_id=None):
|
|
calls.append((file_value, group_id, name))
|
|
|
|
monkeypatch.setattr(gf, "_call_upload", fake_call)
|
|
monkeypatch.setattr(gf, "_s3_url", lambda path, policy: link)
|
|
|
|
file = _make_files(tmp_path / "src", ["a.zip"])[0]
|
|
assert await gf.upload_group_file(file, 123) is True
|
|
assert calls == [(link, 123, "a.zip")]
|
|
|
|
|
|
async def test_s3_call_failure_falls_back_to_local(gf, tmp_path, monkeypatch):
|
|
seen: list[str] = []
|
|
link = "http://192.168.2.15:5246/PLANA/x.zip?sig=1"
|
|
|
|
async def fake_call(file_value, group_id, name, folder_id=None):
|
|
seen.append(file_value)
|
|
if file_value.startswith("http"):
|
|
raise RuntimeError("链接不可达")
|
|
|
|
monkeypatch.setattr(gf, "_call_upload", fake_call)
|
|
monkeypatch.setattr(gf, "_s3_url", lambda path, policy: link)
|
|
|
|
file = _make_files(tmp_path / "src", ["a.zip"])[0]
|
|
assert await gf.upload_group_file(file, 123) is True
|
|
assert seen[0] == link
|
|
assert seen[1] == gf._file_uri(file)
|
|
|
|
|
|
async def test_no_s3_link_falls_back_to_local(gf, tmp_path, monkeypatch):
|
|
seen: list[str] = []
|
|
|
|
async def fake_call(file_value, group_id, name, folder_id=None):
|
|
seen.append(file_value)
|
|
|
|
monkeypatch.setattr(gf, "_call_upload", fake_call)
|
|
monkeypatch.setattr(gf, "_s3_url", lambda path, policy: "")
|
|
|
|
file = _make_files(tmp_path / "src", ["a.zip"])[0]
|
|
assert await gf.upload_group_file(file, 123) is True
|
|
assert seen == [gf._file_uri(file)]
|
|
|
|
|
|
async def test_both_transports_fail(gf, tmp_path, monkeypatch):
|
|
async def always_fail(*args, **kwargs):
|
|
raise RuntimeError("nope")
|
|
|
|
monkeypatch.setattr(gf, "_call_upload", always_fail)
|
|
monkeypatch.setattr(gf, "_s3_url", lambda path, policy: "http://x/y.zip")
|
|
|
|
file = _make_files(tmp_path / "src", ["a.zip"])[0]
|
|
assert await gf.upload_group_file(file, 123) is False
|
|
|
|
|
|
async def test_missing_file_short_circuits(gf, tmp_path, monkeypatch):
|
|
called = []
|
|
monkeypatch.setattr(gf, "_call_upload", lambda *a, **kw: called.append(1))
|
|
assert await gf.upload_group_file(tmp_path / "nope.zip", 123) is False
|
|
assert called == []
|
|
|
|
|
|
# ───────────────────────── 投递编排 ─────────────────────────
|
|
|
|
|
|
def _record_uploads(gf, monkeypatch, path: Path, ok: bool = True) -> list:
|
|
calls: list = []
|
|
|
|
async def fake_upload(file_path, group_id, **kwargs):
|
|
calls.append((str(file_path), group_id))
|
|
return ok
|
|
|
|
monkeypatch.setattr(gf, "upload_group_file", fake_upload)
|
|
return calls
|
|
|
|
|
|
async def test_zip_mode_uploads_single_archive(gf, tmp_path, monkeypatch):
|
|
archive = tmp_path / "pack.zip"
|
|
archive.write_bytes(b"zip")
|
|
calls = _record_uploads(gf, monkeypatch, archive)
|
|
monkeypatch.setattr(
|
|
gf, "build_archive", lambda *a, **kw: archive
|
|
)
|
|
|
|
files = _make_files(tmp_path / "src", ["a.txt", "b.txt"])
|
|
ok = await gf.upload_group_files(files, 123, title="t", password="pw")
|
|
assert ok is True
|
|
assert calls == [(str(archive), 123)]
|
|
|
|
|
|
async def test_raw_mode_uploads_each_file(gf, tmp_path, monkeypatch):
|
|
calls = _record_uploads(gf, monkeypatch, tmp_path)
|
|
files = _make_files(tmp_path / "src", ["a.txt", "b.txt"])
|
|
|
|
ok = await gf.upload_group_files(files, 123, zip_files=False)
|
|
assert ok is True
|
|
assert [c[0] for c in calls] == [str(p) for p in files]
|
|
|
|
|
|
async def test_raw_mode_prefixes_work_name_for_multi_files(gf, tmp_path, monkeypatch):
|
|
"""多图作品逐个传时，001.jpg 这类成员名要带上作品名（单文件不加）"""
|
|
seen: list = []
|
|
|
|
async def fake_upload(file_path, group_id, **kwargs):
|
|
seen.append(kwargs.get("name"))
|
|
return True
|
|
|
|
monkeypatch.setattr(gf, "upload_group_file", fake_upload)
|
|
|
|
images = _make_files(tmp_path / "src", ["001.jpg", "002.jpg"])
|
|
await gf.upload_group_files(
|
|
images, 123, zip_files=False, rel_dir="作者_9/海边日落"
|
|
)
|
|
assert seen == ["海边日落_001.jpg", "海边日落_002.jpg"]
|
|
|
|
seen.clear()
|
|
single = _make_files(tmp_path / "src2", ["作品.mp4"])
|
|
await gf.upload_group_files(single, 123, zip_files=False, rel_dir="作者_9")
|
|
assert seen == ["作品.mp4"] # 单文件时文件名本身就是作品名，不再加前缀
|
|
|
|
|
|
async def test_pack_failure_skips_upload_entirely(gf, tmp_path, monkeypatch):
|
|
"""打包失败（例如缺 pyzipper）时不能退化成上传原文件"""
|
|
calls = _record_uploads(gf, monkeypatch, tmp_path)
|
|
|
|
def boom(*args, **kwargs):
|
|
raise RuntimeError("pyzipper 未安装")
|
|
|
|
monkeypatch.setattr(gf, "build_archive", boom)
|
|
files = _make_files(tmp_path / "src", ["a.txt"])
|
|
|
|
ok = await gf.upload_group_files(files, 123, title="t", password="pw")
|
|
assert ok is False
|
|
assert calls == []
|
|
|
|
|
|
async def test_empty_file_list(gf, tmp_path, monkeypatch):
|
|
calls = _record_uploads(gf, monkeypatch, tmp_path)
|
|
assert await gf.upload_group_files([], 123) is False
|
|
assert calls == []
|