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>
316 lines
12 KiB
Python
316 lines
12 KiB
Python
"""视频解析插件群策略 (policy.py) 单元测试
|
|
|
|
policy.py 只依赖标准库 + nonebot.logger，故用 importlib 按文件路径裸加载
|
|
（与 test_group_daily_history_store.py 同法），避免触发插件包 __init__ 的
|
|
NoneBot 初始化。
|
|
"""
|
|
|
|
import importlib.util
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
_MODULE_PATH = (
|
|
Path(__file__).resolve().parents[1]
|
|
/ "hexi"
|
|
/ "plugins"
|
|
/ "nonebot_plugin_video_analysis"
|
|
/ "policy.py"
|
|
)
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def policy():
|
|
"""以独立模块名加载 policy.py，避免与插件包 __init__ 冲突"""
|
|
spec = importlib.util.spec_from_file_location("video_policy_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 test_normalize_platform_aliases(policy):
|
|
assert policy.normalize_platform("xhs") == "小红书"
|
|
assert policy.normalize_platform("小红书") == "小红书"
|
|
assert policy.normalize_platform("b23") == "哔哩哔哩"
|
|
assert policy.normalize_platform("B站") == "哔哩哔哩"
|
|
assert policy.normalize_platform("yt") == "Youtube"
|
|
assert policy.normalize_platform("twitter") == "X"
|
|
assert policy.normalize_platform(" X ") == "X"
|
|
assert policy.normalize_platform("微博") is None
|
|
assert policy.normalize_platform(None) is None
|
|
|
|
|
|
def test_normalize_platforms_dedupe_and_order(policy):
|
|
# 别名去重 + 按 PLATFORMS 顺序排列（与输入顺序无关）
|
|
assert policy.normalize_platforms(["x", "xhs", "twitter"]) == ["小红书", "X"]
|
|
# 字符串按分隔符拆
|
|
assert policy.normalize_platforms("小红书,xhs 抖音") == ["小红书", "抖音"]
|
|
# 非法值丢弃
|
|
assert policy.normalize_platforms(["微博", "xhs"]) == ["小红书"]
|
|
assert policy.normalize_platforms(None) == []
|
|
assert policy.normalize_platforms(123) == []
|
|
|
|
|
|
def test_apply_platform_diff_incremental(policy):
|
|
# 全部带 +/- → 增量
|
|
assert policy.apply_platform_diff([], ["+xhs"]) == ["小红书"]
|
|
assert policy.apply_platform_diff(["小红书"], ["+X"]) == ["小红书", "X"]
|
|
assert policy.apply_platform_diff(["小红书", "X"], ["-X"]) == ["小红书"]
|
|
assert policy.apply_platform_diff(["小红书"], ["+xhs"]) == ["小红书"] # 去重
|
|
assert policy.apply_platform_diff(["小红书"], ["-X"]) == ["小红书"] # 删不存在的
|
|
assert policy.apply_platform_diff([], ["+xhs", "+b23"]) == ["小红书", "哔哩哔哩"]
|
|
|
|
|
|
def test_apply_platform_diff_replace_and_invalid(policy):
|
|
# 出现裸平台名 → 整体覆盖
|
|
assert policy.apply_platform_diff(["小红书"], ["x"]) == ["X"]
|
|
assert policy.apply_platform_diff(["小红书"], []) is None
|
|
assert policy.apply_platform_diff(["小红书"], ["微博"]) is None
|
|
assert policy.apply_platform_diff(["小红书"], ["+微博"]) is None
|
|
|
|
|
|
def test_match_platform_domains(policy):
|
|
assert policy.match_platform("https://x.com/a/status/1") == "X"
|
|
assert policy.match_platform("https://twitter.com/a/status/1") == "X"
|
|
assert policy.match_platform("https://b23.tv/abc") == "哔哩哔哩"
|
|
assert policy.match_platform("https://v.douyin.com/abc") == "抖音"
|
|
assert policy.match_platform("https://youtu.be/abc") == "Youtube"
|
|
assert policy.match_platform("https://example.com/a.mp4") is None
|
|
|
|
|
|
# ───────────────────────── 策略模型 ─────────────────────────
|
|
|
|
|
|
def test_policy_from_dict_sanitizes(policy):
|
|
p = policy.Policy.from_dict(
|
|
{
|
|
"auto": 1,
|
|
"auto_link": ["xhs", "xhs", "微博"],
|
|
"ban_link": ["twitter"],
|
|
"plan": "b",
|
|
"upload_public": True,
|
|
"send_link": True,
|
|
"upload_group_file": True,
|
|
"unknown_key": "ignored",
|
|
}
|
|
)
|
|
assert p.auto is True
|
|
assert p.auto_link == ["小红书"]
|
|
assert p.ban_link == ["X"]
|
|
assert p.plan == "B"
|
|
assert p.to_dict()["upload_group_file"] is True
|
|
assert "unknown_key" not in p.to_dict()
|
|
|
|
|
|
def test_policy_from_dict_bad_values_fall_back(policy):
|
|
p = policy.Policy.from_dict({"plan": "Z", "auto_link": "b23"})
|
|
assert p.plan == "C" # 非法 plan 回退默认
|
|
assert p.auto_link == ["哔哩哔哩"]
|
|
# 空/非 dict 输入不炸
|
|
assert policy.Policy.from_dict(None) == policy.Policy()
|
|
assert policy.Policy.from_dict([1, 2]) == policy.Policy()
|
|
|
|
|
|
def test_sends_link_requires_public(policy):
|
|
assert policy.Policy(upload_public=True, send_link=True).sends_link is True
|
|
# 没开公网就没有链接可发
|
|
assert policy.Policy(upload_public=False, send_link=True).sends_link is False
|
|
|
|
|
|
def test_allows_group_file(policy):
|
|
off = policy.Policy(upload_group_file=False)
|
|
assert off.allows_group_file("抖音") is False
|
|
assert off.allows_group_file(None) is False
|
|
|
|
# 开了但没限定平台 → 所有平台（含识别不出的）都传
|
|
every = policy.Policy(upload_group_file=True)
|
|
assert every.allows_group_file("抖音") is True
|
|
assert every.allows_group_file(None) is True
|
|
|
|
# 限定了平台 → 只传清单里的，识别不出平台时按"不在清单里"处理
|
|
only_douyin = policy.Policy(
|
|
upload_group_file=True, group_file_platforms=["抖音"]
|
|
)
|
|
assert only_douyin.allows_group_file("抖音") is True
|
|
assert only_douyin.allows_group_file("X") is False
|
|
assert only_douyin.allows_group_file(None) is False
|
|
|
|
|
|
def test_policy_construct_normalizes(policy):
|
|
"""直接构造 Policy(...) 也要归一：绕过 from_dict 传别名不该静默失效"""
|
|
p = policy.Policy(
|
|
plan="z",
|
|
auto_link=["xhs"],
|
|
upload_group_file=True,
|
|
group_file_platforms=["douyin"],
|
|
)
|
|
assert p.plan == "C"
|
|
assert p.auto_link == ["小红书"]
|
|
assert p.group_file_platforms == ["抖音"]
|
|
assert p.allows_group_file("抖音") is True
|
|
|
|
|
|
def test_group_file_platforms_normalized(policy):
|
|
p = policy.Policy.from_dict(
|
|
{"group_file_platforms": ["douyin", "b23", "微博"], "upload_group_file": True}
|
|
)
|
|
# 别名归一成规范标签、非法值丢弃、顺序按 PLATFORMS
|
|
assert p.group_file_platforms == ["哔哩哔哩", "抖音"]
|
|
|
|
|
|
def test_banned_and_auto_matched(policy):
|
|
p = policy.Policy(auto_link=["小红书"], ban_link=["X"])
|
|
assert p.auto_matched("https://xhslink.com/a") == "小红书"
|
|
assert p.auto_matched("https://x.com/a") is None
|
|
assert p.banned("https://x.com/a") == "X"
|
|
assert p.banned("https://twitter.com/a") == "X"
|
|
assert p.banned("https://douyin.com/a") is None
|
|
|
|
|
|
# ───────────────────────── 迁移 ─────────────────────────
|
|
|
|
|
|
def test_migrate_v1_to_v3(policy):
|
|
data, changed = policy.migrate(
|
|
{
|
|
"WHITE_LIST": ["111", "222"],
|
|
"AUTO_ANALYSIS": ["111"],
|
|
"PLANA": ["111"],
|
|
"PLANB": ["222"],
|
|
"BLACK_LIST": ["999"],
|
|
}
|
|
)
|
|
assert changed is True
|
|
assert set(data["groups"]) == {"111", "222"}
|
|
# v1 的 plan=B 隐含公网，迁移后显式补上
|
|
assert data["groups"]["222"]["upload_public"] is True
|
|
assert data["groups"]["111"]["plan"] == "A"
|
|
assert data["groups"]["111"]["auto"] is True
|
|
assert data["blacklist"] == ["999"]
|
|
assert data["default"] == policy.Policy().to_dict()
|
|
|
|
|
|
def test_migrate_v2_plan_b_gets_public(policy):
|
|
data, changed = policy.migrate(
|
|
{
|
|
"groups": {
|
|
"111": {"auto": True, "plan": "B"},
|
|
"222": {"plan": "C"},
|
|
"333": {},
|
|
},
|
|
"blacklist": [],
|
|
}
|
|
)
|
|
assert changed is True
|
|
assert data["groups"]["111"]["upload_public"] is True
|
|
assert data["groups"]["222"]["upload_public"] is False
|
|
assert data["groups"]["333"]["auto"] is False
|
|
|
|
|
|
def test_migrate_keeps_explicit_public_flag(policy):
|
|
# 显式写了 upload_public 的条目不被迁移覆盖
|
|
data, _ = policy.migrate(
|
|
{"groups": {"111": {"plan": "B", "upload_public": False}}, "blacklist": []}
|
|
)
|
|
assert data["groups"]["111"]["upload_public"] is False
|
|
|
|
|
|
def test_migrate_idempotent(policy):
|
|
first, _ = policy.migrate({"groups": {"111": {"plan": "B"}}, "blacklist": []})
|
|
second, changed = policy.migrate(first)
|
|
assert changed is False
|
|
assert second == first
|
|
|
|
|
|
# ───────────────────────── 存储 ─────────────────────────
|
|
|
|
|
|
async def test_store_update_and_persist(policy, tmp_path):
|
|
path = tmp_path / "list.json"
|
|
store = policy.PolicyStore(path)
|
|
store.load()
|
|
assert path.exists() # 首次 load 落盘
|
|
|
|
await store.update_group("111", create=True, auto=True, plan="A")
|
|
await store.update_group("222", create=True, auto_link=["xhs"])
|
|
|
|
# 内存读
|
|
assert store.get("111").auto is True
|
|
assert store.get("222").auto_link == ["小红书"]
|
|
# 未配置的群回落 default
|
|
assert store.get("999") == store.default_policy()
|
|
assert store.is_whitelisted("111") is True
|
|
assert store.is_whitelisted("999") is False
|
|
|
|
on_disk = json.loads(path.read_text(encoding="utf-8"))
|
|
assert on_disk["groups"]["111"]["auto"] is True
|
|
assert on_disk["groups"]["222"]["auto_link"] == ["小红书"]
|
|
assert not path.with_name("list.json.tmp").exists()
|
|
|
|
|
|
async def test_store_update_requires_whitelist_unless_create(policy, tmp_path):
|
|
store = policy.PolicyStore(tmp_path / "list.json")
|
|
assert await store.update_group("111", auto=True) is None
|
|
assert store.is_whitelisted("111") is False
|
|
|
|
|
|
async def test_store_remove_group_and_blacklist(policy, tmp_path):
|
|
store = policy.PolicyStore(tmp_path / "list.json")
|
|
await store.update_group("111", create=True)
|
|
assert await store.remove_group("111") is True
|
|
assert await store.remove_group("111") is False
|
|
assert store.is_whitelisted("111") is False
|
|
|
|
assert await store.add_blacklist(999) is True
|
|
assert await store.add_blacklist("999") is False # 幂等（int/str 同一个人）
|
|
assert store.is_blacklisted(999) is True
|
|
assert store.blacklist() == ["999"]
|
|
assert await store.remove_blacklist("999") is True
|
|
assert store.blacklist() == []
|
|
|
|
|
|
async def test_store_concurrent_updates_keep_all(policy, tmp_path):
|
|
"""并发写不能丢群：每次写都从事件循环线程快照，写线程不碰共享状态"""
|
|
path = tmp_path / "list.json"
|
|
store = policy.PolicyStore(path)
|
|
store.load()
|
|
|
|
import asyncio
|
|
|
|
await asyncio.gather(
|
|
*(store.update_group(str(1000 + i), create=True, auto=True) for i in range(30))
|
|
)
|
|
|
|
on_disk = json.loads(path.read_text(encoding="utf-8"))
|
|
assert len(on_disk["groups"]) == 30
|
|
assert len(store.all_groups()) == 30
|
|
|
|
|
|
async def test_store_persists_group_file_platforms(policy, tmp_path):
|
|
"""POLICY_FIELDS 里漏了新字段的话这次更新会被静默丢掉"""
|
|
path = tmp_path / "list.json"
|
|
store = policy.PolicyStore(path)
|
|
await store.update_group(
|
|
"111", create=True, upload_group_file=True, group_file_platforms=["xhs"]
|
|
)
|
|
assert store.get("111").group_file_platforms == ["小红书"]
|
|
on_disk = json.loads(path.read_text(encoding="utf-8"))
|
|
assert on_disk["groups"]["111"]["group_file_platforms"] == ["小红书"]
|
|
assert on_disk["groups"]["111"]["upload_group_file"] is True
|
|
|
|
|
|
async def test_store_default_policy_edit(policy, tmp_path):
|
|
store = policy.PolicyStore(tmp_path / "list.json")
|
|
await store.set_default(policy.Policy(plan="B", upload_public=True))
|
|
assert store.default_policy().plan == "B"
|
|
assert store.default_policy().upload_public is True
|
|
# 群未配置时读到的就是 default
|
|
assert store.get("111").plan == "B"
|