feat(cherry-tomato): 新增图片混淆/解混淆插件（小番茄）
Gilbert 曲线像素重排，可与网页版互解；`混淆` / `解混淆` 命令 （回复图片或图文同条消息发送），附 tests/test_cherry_tomato.py Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
sansenhoshiandClaude Code committed
1 parent
968b3d1cfe
commit
fb15b976aa
9 files changed
+862
No files matched your search
@@ -0,0 +1,22 @@
|
||||
"""图片混淆（小番茄）—— Gilbert 曲线像素重排，可与网页版互解。"""
|
||||
|
||||
from nonebot import require
|
||||
from nonebot.plugin import PluginMetadata
|
||||
|
||||
require("nonebot_plugin_alconna")
|
||||
|
||||
__plugin_meta__ = PluginMetadata(
|
||||
name="图片混淆(小番茄)",
|
||||
description=(
|
||||
"小番茄图片混淆：Gilbert 曲线 + 黄金比例偏移做像素重排，"
|
||||
"混淆/解混淆互为逆操作，可与网页版(hideImg)互解"
|
||||
),
|
||||
usage="回复一张图片发送「混淆」或「解混淆」（图片和命令同一条消息发也行）",
|
||||
type="application",
|
||||
)
|
||||
|
||||
# 入口只做四件事：require / meta / 导入子模块 / 配置注册
|
||||
from . import config as _config # noqa: E402,F401
|
||||
from . import handlers as _handlers # noqa: E402,F401
|
||||
|
||||
_config.register_config()
|
||||
@@ -0,0 +1,84 @@
|
||||
"""配置：按《插件配置文件标准》注册到统一 Web 管理台。
|
||||
|
||||
3 个配置项来源都是本模块常量：Web 保存后由 register_config_items 的
|
||||
setter（setattr 本模块）热更新、立即生效；重启后按 plugin_config.json
|
||||
已存值回退（读取一律走本模块 helper，不要在别处缓存值）。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import sys as _sys
|
||||
from typing import Any
|
||||
|
||||
from hexi.web_hub.config_standard import register_config_items
|
||||
from hexi.web_hub.web_config import get_effective_value
|
||||
|
||||
_PLUGIN_ID = __package__ or __name__.rsplit(".", 1)[0]
|
||||
|
||||
# —— 可配置常量（来源为本模块） ——
|
||||
# 加密前超过该像素数则等比缩小（并在回复里提示已压缩）
|
||||
MAX_PIXELS = 8_000_000
|
||||
# 解混淆硬上限：解密不能缩放（缩了就永远解不出），超过直接拒绝
|
||||
DECRYPT_MAX_PIXELS = 16_000_000
|
||||
# 输出 JPEG 质量（1-95，网页版固定最高档 95）
|
||||
JPEG_QUALITY = 95
|
||||
|
||||
|
||||
def _as_int(key: str, default: int) -> int:
|
||||
value: Any = get_effective_value(_PLUGIN_ID, key, default)
|
||||
if value in (None, ""):
|
||||
return default
|
||||
try:
|
||||
return int(value)
|
||||
except (TypeError, ValueError):
|
||||
return default
|
||||
|
||||
|
||||
def max_pixels() -> int:
|
||||
"""加密缩放阈值（像素数）。"""
|
||||
return _as_int("MAX_PIXELS", MAX_PIXELS)
|
||||
|
||||
|
||||
def decrypt_max_pixels() -> int:
|
||||
"""解混淆硬上限（像素数）。"""
|
||||
return _as_int("DECRYPT_MAX_PIXELS", DECRYPT_MAX_PIXELS)
|
||||
|
||||
|
||||
def jpeg_quality() -> int:
|
||||
"""输出 JPEG 质量。"""
|
||||
return _as_int("JPEG_QUALITY", JPEG_QUALITY)
|
||||
|
||||
|
||||
def register_config() -> None:
|
||||
register_config_items(
|
||||
_PLUGIN_ID,
|
||||
[
|
||||
{
|
||||
"key": "MAX_PIXELS",
|
||||
"label": "加密最大像素数",
|
||||
"type": "int",
|
||||
"default": MAX_PIXELS,
|
||||
"description": (
|
||||
"混淆前超过该像素数则等比缩小（回复里会提示已压缩）。"
|
||||
"默认 8000000（8MP）"
|
||||
),
|
||||
},
|
||||
{
|
||||
"key": "DECRYPT_MAX_PIXELS",
|
||||
"label": "解混淆像素上限",
|
||||
"type": "int",
|
||||
"default": DECRYPT_MAX_PIXELS,
|
||||
"description": (
|
||||
"解混淆不能缩放，超过该像素数直接拒绝。默认 16000000（16MP）"
|
||||
),
|
||||
},
|
||||
{
|
||||
"key": "JPEG_QUALITY",
|
||||
"label": "输出 JPEG 质量",
|
||||
"type": "int",
|
||||
"default": JPEG_QUALITY,
|
||||
"description": "1-95，默认 95（与网页版一致的最高档）。",
|
||||
},
|
||||
],
|
||||
store=_sys.modules[__name__],
|
||||
)
|
||||
@@ -0,0 +1,3 @@
|
||||
"""小番茄图片混淆 —— Trigger 层。"""
|
||||
|
||||
from .obfuscate import deobf_cmd, obf_cmd # noqa: F401
|
||||
@@ -0,0 +1,122 @@
|
||||
"""混淆/解混淆 —— 触发器（Trigger）层。
|
||||
|
||||
回复一张图片后发「混淆」或「解混淆」触发（图片与命令同一条消息直接发也行，
|
||||
不需要 @bot）；下载图片 → 变换 → 结果图发回。算法层见 utils/obfuscation.py。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
|
||||
from nonebot import on_command
|
||||
from nonebot.adapters.onebot.v11 import Bot, MessageEvent
|
||||
from nonebot.log import logger
|
||||
from nonebot_plugin_alconna import UniMessage
|
||||
|
||||
from hexi.core.cooldown import cooldown
|
||||
from hexi.core.message_utils import get_reply_message
|
||||
|
||||
from .. import config
|
||||
from ..services.pipeline import download_image, run_pipeline
|
||||
|
||||
# 两个命令共用同一冷却器（name 相同）：同一用户 10 秒内不能连发混淆/解混淆
|
||||
obf_cmd = cooldown(10, hint="处理太频繁了，{secs} 秒后再试", name="cherry_tomato")(
|
||||
on_command("混淆", aliases={"小番茄混淆"}, priority=10, block=True)
|
||||
)
|
||||
deobf_cmd = cooldown(10, hint="处理太频繁了，{secs} 秒后再试", name="cherry_tomato")(
|
||||
on_command("解混淆", aliases={"小番茄解混淆"}, priority=10, block=True)
|
||||
)
|
||||
|
||||
_MD5_RE = re.compile(r"[0-9A-Fa-f]{32}")
|
||||
|
||||
|
||||
@obf_cmd.handle()
|
||||
async def _obfuscate(event: MessageEvent, bot: Bot) -> None:
|
||||
await _handle(event, bot, decrypt=False)
|
||||
|
||||
|
||||
@deobf_cmd.handle()
|
||||
async def _deobfuscate(event: MessageEvent, bot: Bot) -> None:
|
||||
await _handle(event, bot, decrypt=True)
|
||||
|
||||
|
||||
async def _extract_image(
|
||||
event: MessageEvent, bot: Bot
|
||||
) -> tuple[str | None, str | None]:
|
||||
"""取被回复（或当前）消息里第一张图的 (url, file)。
|
||||
|
||||
优先段里的 url；没有 url 时用 file 调 OneBot get_image 现换一个。
|
||||
file 仅用于 offpic 链接失效时的 md5 兜底重试。
|
||||
"""
|
||||
for seg in get_reply_message(event):
|
||||
if seg.type != "image":
|
||||
continue
|
||||
data = seg.data or {}
|
||||
url = data.get("url")
|
||||
if url:
|
||||
return str(url), str(data.get("file") or "")
|
||||
file = data.get("file")
|
||||
if not file:
|
||||
continue
|
||||
try:
|
||||
info = await bot.get_image(file=str(file))
|
||||
except Exception as e: # noqa: BLE001
|
||||
logger.warning(f"cherry_tomato: get_image 取 url 失败: {e!r}")
|
||||
continue
|
||||
url = info.get("url") if isinstance(info, dict) else None
|
||||
if url:
|
||||
return str(url), str(file)
|
||||
return None, None
|
||||
|
||||
|
||||
def _offpic_md5(url: str, file: str | None) -> str | None:
|
||||
"""从 file 字段或 offpic 链接里找 32 位十六进制（md5）串；找不到返回 None。"""
|
||||
for candidate in (file, url):
|
||||
match = _MD5_RE.search(candidate or "")
|
||||
if match:
|
||||
return match.group(0).upper()
|
||||
return None
|
||||
|
||||
|
||||
async def _download_with_fallback(url: str, file: str | None) -> bytes | None:
|
||||
"""下载图片；offpic 老链接失败时用 md5 换 gchat 链接重试一次（picfinder 先例）。"""
|
||||
data = await download_image(url)
|
||||
if data is not None:
|
||||
return data
|
||||
if "c2cpicdw.qpic.cn/offpic_new/" in url:
|
||||
md5 = _offpic_md5(url, file)
|
||||
if md5:
|
||||
retry_url = f"http://gchat.qpic.cn/gchatpic_new/0/0-0-{md5}/0?term=2"
|
||||
logger.info("cherry_tomato: offpic 原链下载失败，改用 gchat 链接重试")
|
||||
data = await download_image(retry_url)
|
||||
return data
|
||||
|
||||
|
||||
async def _handle(event: MessageEvent, bot: Bot, *, decrypt: bool) -> None:
|
||||
action = "解混淆" if decrypt else "混淆"
|
||||
|
||||
url, file = await _extract_image(event, bot)
|
||||
if not url:
|
||||
await UniMessage.text(
|
||||
"请回复一张图片再发这个命令，或把图片和命令一起发~"
|
||||
).finish()
|
||||
|
||||
data = await _download_with_fallback(url, file)
|
||||
if data is None:
|
||||
logger.warning(f"cherry_tomato: 图片下载失败 url={url}")
|
||||
await UniMessage.text("图片下载失败，可能链接已过期，重发一次试试").finish()
|
||||
|
||||
logger.info(f"cherry_tomato: {action}开始 user={event.get_user_id()} url={url}")
|
||||
max_pixels = config.decrypt_max_pixels() if decrypt else config.max_pixels()
|
||||
out, err, notes = await run_pipeline(
|
||||
data, decrypt=decrypt, max_pixels=max_pixels, quality=config.jpeg_quality()
|
||||
)
|
||||
if out is None:
|
||||
logger.warning(f"cherry_tomato: {action}失败: {err}")
|
||||
await UniMessage.text(err or f"{action}失败了，换一张图试试").finish()
|
||||
|
||||
image = UniMessage.image(raw=out)
|
||||
if notes:
|
||||
await (UniMessage.text("；".join(notes) + "\n") + image).send()
|
||||
else:
|
||||
await image.send()
|
||||
@@ -0,0 +1,3 @@
|
||||
"""小番茄图片混淆 —— Service 层。"""
|
||||
|
||||
from .pipeline import download_image, run_pipeline # noqa: F401
|
||||
@@ -0,0 +1,69 @@
|
||||
"""小番茄图片混淆 —— 编排（Service 层）。无 nonebot 依赖：下载 / 解码 / 变换 / 编码。"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import io
|
||||
|
||||
import httpx
|
||||
from PIL import Image
|
||||
|
||||
from ..utils.obfuscation import obfuscate, to_jpeg_bytes
|
||||
|
||||
|
||||
async def download_image(url: str) -> bytes | None:
|
||||
"""下载图片；失败（超时/404/断链）返回 None，由调用方决定提示与重试。"""
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=30) as client:
|
||||
resp = await client.get(url)
|
||||
resp.raise_for_status()
|
||||
return resp.content
|
||||
except Exception: # noqa: BLE001 - 网络失败一律走 None，交给上层提示
|
||||
return None
|
||||
|
||||
|
||||
async def run_pipeline(
|
||||
image_bytes: bytes,
|
||||
*,
|
||||
decrypt: bool,
|
||||
max_pixels: int,
|
||||
quality: int,
|
||||
) -> tuple[bytes | None, str | None, list[str]]:
|
||||
"""PIL 解码 → 混淆/解混淆（线程池） → JPEG 编码。
|
||||
|
||||
max_pixels 语义随方向不同：
|
||||
- 加密：超过则等比缩小（与网页版一致，自洽即可）；
|
||||
- 解密：不能缩放（缩了就永远解不出），超过直接拒绝。
|
||||
|
||||
Returns: (jpeg_bytes, err, notes) —— err 非空时 bytes 为 None；
|
||||
notes 是给用户的附加提示（已压缩 / 动图仅首帧）。
|
||||
"""
|
||||
try:
|
||||
img = Image.open(io.BytesIO(image_bytes))
|
||||
img.load() # 提前解码，截断/坏文件在这里就暴露
|
||||
except Exception: # noqa: BLE001
|
||||
return None, "打不开这张图，可能不是有效的图片文件", []
|
||||
|
||||
width, height = img.size
|
||||
if decrypt and width * height > max_pixels:
|
||||
return (
|
||||
None,
|
||||
f"图片太大（{width}x{height}），解混淆不能缩放"
|
||||
f"（上限 {max_pixels / 1_000_000:.0f}MP），先裁剪或压小再试",
|
||||
[],
|
||||
)
|
||||
|
||||
try:
|
||||
out, scaled, extra_frames = await asyncio.to_thread(
|
||||
obfuscate, img, decrypt=decrypt, max_pixels=max_pixels
|
||||
)
|
||||
data = await asyncio.to_thread(to_jpeg_bytes, out, quality)
|
||||
except Exception: # noqa: BLE001
|
||||
return None, "处理这张图失败了，换一张再试试", []
|
||||
|
||||
notes: list[str] = []
|
||||
if scaled:
|
||||
notes.append(f"图片超过 {max_pixels / 1_000_000:.0f}MP，已等比缩小后再混淆")
|
||||
if extra_frames:
|
||||
notes.append("这是动图，只处理了第一帧")
|
||||
return data, None, notes
|
||||
@@ -0,0 +1,10 @@
|
||||
"""小番茄图片混淆 —— 纯工具层（无 nonebot 依赖，可单测）。"""
|
||||
|
||||
from .obfuscation import ( # noqa: F401
|
||||
decrypt_pixels,
|
||||
encrypt_pixels,
|
||||
fit_within,
|
||||
gilbert_offset,
|
||||
obfuscate,
|
||||
to_jpeg_bytes,
|
||||
)
|
||||
@@ -0,0 +1,257 @@
|
||||
"""小番茄图片混淆 —— 纯算法层（无 nonebot / 无配置依赖，可裸加载单测）。
|
||||
|
||||
算法与 singularpoint.cn/hideImg1.html、GitHub ok8634673/Cherry-tomato-image-obfuscation
|
||||
一致：Gilbert（广义 Hilbert）空间填充曲线 + 黄金比例偏移做像素重排，混淆/解混淆
|
||||
互为逆操作，混淆结果可与网页版互解（关键：曲线与偏移逐像素一致、尺寸不变）。
|
||||
|
||||
性能：朴素实现要逐像素 append 出 total 个坐标（12MP 约 17s）。这里改为「小块模板
|
||||
缓存」：曲线递归到 w*h <= _TILE_AREA 的轴对齐子矩形时不再下拆，直接拼接该
|
||||
「形状+朝向」的相对坐标模板（模板本身用参考递归生成 + lru_cache，共 8 种朝向），
|
||||
每个模板实例只做一条向量化赋值。1920x1080 生成位置数组约 0.03s。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import functools
|
||||
import io
|
||||
import math
|
||||
|
||||
import numpy as np
|
||||
from PIL import Image, ImageOps
|
||||
|
||||
# 递归到该面积的子矩形改用缓存模板（模板数量与命中率的折中）
|
||||
_TILE_AREA = 1024
|
||||
# 模板缓存条数：形状+朝向空间有限，512 条足够任何尺寸复用
|
||||
_TILE_CACHE_SIZE = 512
|
||||
|
||||
|
||||
def _sign(x: int) -> int:
|
||||
if x > 0:
|
||||
return 1
|
||||
if x < 0:
|
||||
return -1
|
||||
return 0
|
||||
|
||||
|
||||
def _collect(x, y, ax, ay, bx, by, coords) -> None:
|
||||
"""参考实现（gilbert2d）的递归本体，逐点收集坐标（仅用于生成模板）。
|
||||
|
||||
与参考实现逐行一致；子问题恒为轴对齐矩形（a、b 各沿一轴），起点平移到
|
||||
(0, 0) 后形状不变，故可用「(w,h)+方向」作模板键。
|
||||
"""
|
||||
w = abs(ax + ay)
|
||||
h = abs(bx + by)
|
||||
|
||||
dax = _sign(ax)
|
||||
day = _sign(ay)
|
||||
dbx = _sign(bx)
|
||||
dby = _sign(by)
|
||||
|
||||
if h == 1:
|
||||
for _ in range(w):
|
||||
coords.append((x, y))
|
||||
x += dax
|
||||
y += day
|
||||
return
|
||||
|
||||
if w == 1:
|
||||
for _ in range(h):
|
||||
coords.append((x, y))
|
||||
x += dbx
|
||||
y += dby
|
||||
return
|
||||
|
||||
ax2 = ax // 2
|
||||
ay2 = ay // 2
|
||||
bx2 = bx // 2
|
||||
by2 = by // 2
|
||||
|
||||
w2 = abs(ax2 + ay2)
|
||||
h2 = abs(bx2 + by2)
|
||||
|
||||
if 2 * w > 3 * h:
|
||||
if (w2 % 2) and (w > 2):
|
||||
ax2 += dax
|
||||
ay2 += day
|
||||
_collect(x, y, ax2, ay2, bx, by, coords)
|
||||
_collect(x + ax2, y + ay2, ax - ax2, ay - ay2, bx, by, coords)
|
||||
else:
|
||||
if (h2 % 2) and (h > 2):
|
||||
bx2 += dbx
|
||||
by2 += dby
|
||||
_collect(x, y, bx2, by2, ax2, ay2, coords)
|
||||
_collect(x + bx2, y + by2, ax, ay, bx - bx2, by - by2, coords)
|
||||
_collect(
|
||||
x + (ax - dax) + (bx2 - dbx),
|
||||
y + (ay - day) + (by2 - dby),
|
||||
-bx2,
|
||||
-by2,
|
||||
-(ax - ax2),
|
||||
-(ay - ay2),
|
||||
coords,
|
||||
)
|
||||
|
||||
|
||||
@functools.lru_cache(maxsize=_TILE_CACHE_SIZE)
|
||||
def _tile(
|
||||
w: int, h: int, dax: int, day: int, dbx: int, dby: int
|
||||
) -> tuple[np.ndarray, np.ndarray]:
|
||||
"""生成「a 长 w 朝 (dax,day)、b 长 h 朝 (dbx,dby)」子矩形的相对坐标模板。
|
||||
|
||||
返回 (lx, ly) 两个 int32 数组。起点取 (0,0)，与参考递归逐点一致。
|
||||
"""
|
||||
coords: list[tuple[int, int]] = []
|
||||
_collect(0, 0, w * dax, w * day, h * dbx, h * dby, coords)
|
||||
lx = np.fromiter((c[0] for c in coords), dtype=np.int32, count=len(coords))
|
||||
ly = np.fromiter((c[1] for c in coords), dtype=np.int32, count=len(coords))
|
||||
return lx, ly
|
||||
|
||||
|
||||
def _emit(x, y, ax, ay, bx, by, pos: np.ndarray, k: int, width: int) -> int:
|
||||
"""把子矩形 (x,y,ax,ay,bx,by) 的曲线坐标写入 pos[k:]，返回新的写入位置。
|
||||
|
||||
结构与参考递归完全相同，只在叶子处换用模板 + 一条向量化赋值：
|
||||
相对坐标 (lx,ly) 平移到 (x+lx, y+ly)，展平下标 = (x+y*width) + lx + ly*width。
|
||||
"""
|
||||
w = abs(ax + ay)
|
||||
h = abs(bx + by)
|
||||
|
||||
if h == 1 or w == 1 or w * h <= _TILE_AREA:
|
||||
lx, ly = _tile(w, h, _sign(ax), _sign(ay), _sign(bx), _sign(by))
|
||||
n = lx.shape[0]
|
||||
pos[k : k + n] = (x + y * width) + lx + ly * width
|
||||
return k + n
|
||||
|
||||
dax = _sign(ax)
|
||||
day = _sign(ay)
|
||||
dbx = _sign(bx)
|
||||
dby = _sign(by)
|
||||
|
||||
ax2 = ax // 2
|
||||
ay2 = ay // 2
|
||||
bx2 = bx // 2
|
||||
by2 = by // 2
|
||||
|
||||
w2 = abs(ax2 + ay2)
|
||||
h2 = abs(bx2 + by2)
|
||||
|
||||
if 2 * w > 3 * h:
|
||||
if (w2 % 2) and (w > 2):
|
||||
ax2 += dax
|
||||
ay2 += day
|
||||
k = _emit(x, y, ax2, ay2, bx, by, pos, k, width)
|
||||
k = _emit(x + ax2, y + ay2, ax - ax2, ay - ay2, bx, by, pos, k, width)
|
||||
else:
|
||||
if (h2 % 2) and (h > 2):
|
||||
bx2 += dbx
|
||||
by2 += dby
|
||||
k = _emit(x, y, bx2, by2, ax2, ay2, pos, k, width)
|
||||
k = _emit(x + bx2, y + by2, ax, ay, bx - bx2, by - by2, pos, k, width)
|
||||
k = _emit(
|
||||
x + (ax - dax) + (bx2 - dbx),
|
||||
y + (ay - day) + (by2 - dby),
|
||||
-bx2,
|
||||
-by2,
|
||||
-(ax - ax2),
|
||||
-(ay - ay2),
|
||||
pos,
|
||||
k,
|
||||
width,
|
||||
)
|
||||
return k
|
||||
|
||||
|
||||
def _gilbert_positions(width: int, height: int) -> np.ndarray:
|
||||
"""整图 Gilbert 曲线的展平像素下标序列：pos[i] = 曲线第 i 步的 (x+y*width)。
|
||||
|
||||
是 0..total-1 的一个排列；int32 足够（total < 2^31 由调用方上限保证）。
|
||||
"""
|
||||
total = width * height
|
||||
if total >= 2**31:
|
||||
raise ValueError(f"图片像素数过大：{width}x{height}")
|
||||
pos = np.empty(total, dtype=np.int32)
|
||||
if width >= height:
|
||||
n = _emit(0, 0, width, 0, 0, height, pos, 0, width)
|
||||
else:
|
||||
n = _emit(0, 0, 0, height, width, 0, pos, 0, width)
|
||||
if n != total: # pragma: no cover - 模板/递归出错时的兜底自检
|
||||
raise RuntimeError(f"Gilbert 曲线生成异常：期望 {total} 点，实际 {n}")
|
||||
return pos
|
||||
|
||||
|
||||
def gilbert_offset(total: int) -> int:
|
||||
"""黄金比例偏移，与参考实现（JS: Math.round((√5-1)/2*W*H)）一致。
|
||||
|
||||
与 JS Math.round 的差异只存在于不可能精确命中的 .5 边界，忽略。
|
||||
"""
|
||||
return round((math.sqrt(5) - 1) / 2 * total)
|
||||
|
||||
|
||||
def encrypt_pixels(src: np.ndarray, width: int, height: int) -> np.ndarray:
|
||||
"""混淆：(N,3) RGB 数组重排，dst[curve[i+offset]] = src[curve[i]]。
|
||||
|
||||
以位置数组 pos 数组化：dst[np.roll(pos, -offset)] = src[pos]。
|
||||
"""
|
||||
pos = _gilbert_positions(width, height)
|
||||
dst = np.empty_like(src)
|
||||
dst[np.roll(pos, -gilbert_offset(width * height))] = src[pos]
|
||||
return dst
|
||||
|
||||
|
||||
def decrypt_pixels(src: np.ndarray, width: int, height: int) -> np.ndarray:
|
||||
"""解混淆：encrypt 的逆操作，dst[curve[i]] = src[curve[i+offset]]。"""
|
||||
pos = _gilbert_positions(width, height)
|
||||
dst = np.empty_like(src)
|
||||
dst[pos] = src[np.roll(pos, -gilbert_offset(width * height))]
|
||||
return dst
|
||||
|
||||
|
||||
def fit_within(img: Image.Image, max_pixels: int) -> Image.Image:
|
||||
"""超过 max_pixels 时等比缩小（长宽至少 1 像素）；未超限原样返回。"""
|
||||
w, h = img.size
|
||||
if w * h <= max_pixels:
|
||||
return img
|
||||
scale = math.sqrt(max_pixels / (w * h))
|
||||
new_size = (max(1, int(w * scale)), max(1, int(h * scale)))
|
||||
return img.resize(new_size, Image.Resampling.LANCZOS)
|
||||
|
||||
|
||||
def obfuscate(
|
||||
img: Image.Image, *, decrypt: bool, max_pixels: int
|
||||
) -> tuple[Image.Image, bool, bool]:
|
||||
"""整图混淆/解混淆。
|
||||
|
||||
- 先 ImageOps.exif_transpose 把手机照片的 EXIF 旋转规范化到显示方向
|
||||
（否则输出丢 EXIF 后显示方向会突变），再 convert("RGB")；
|
||||
- 多帧图（GIF/动图）只取第一帧，返回 extra_frames=True 供提示；
|
||||
- 加密超过 max_pixels 等比缩小（scaled=True）；解密绝不缩放（调用方限硬上限）；
|
||||
- 输出是全新 RGB 图，不带 EXIF。
|
||||
|
||||
Returns: (结果图, 是否缩放, 是否丢弃了多余帧)
|
||||
"""
|
||||
extra_frames = getattr(img, "n_frames", 1) > 1
|
||||
# in_place=False 时 exif_transpose 不会返回 None，判空只为过类型检查
|
||||
transposed = ImageOps.exif_transpose(img)
|
||||
img = (transposed or img).convert("RGB")
|
||||
|
||||
scaled = False
|
||||
if not decrypt:
|
||||
before = img.width * img.height
|
||||
img = fit_within(img, max_pixels)
|
||||
scaled = img.width * img.height < before
|
||||
|
||||
width, height = img.size
|
||||
src = np.asarray(img, dtype=np.uint8).reshape(-1, 3)
|
||||
if decrypt:
|
||||
dst = decrypt_pixels(src, width, height)
|
||||
else:
|
||||
dst = encrypt_pixels(src, width, height)
|
||||
out = Image.fromarray(dst.reshape(height, width, 3), "RGB")
|
||||
return out, scaled, extra_frames
|
||||
|
||||
|
||||
def to_jpeg_bytes(img: Image.Image, quality: int) -> bytes:
|
||||
"""按给定质量编码为 JPEG（网页版输出也是 JPEG，质量默认最高档）。"""
|
||||
buf = io.BytesIO()
|
||||
img.save(buf, format="JPEG", quality=quality)
|
||||
return buf.getvalue()
|
||||
@@ -0,0 +1,292 @@
|
||||
"""小番茄图片混淆 —— 纯算法测试。
|
||||
|
||||
utils/obfuscation.py 无 nonebot 依赖，用 importlib 按文件路径裸加载
|
||||
（先例 test_rate_limit.py），避免拉起 NoneBot 运行时。
|
||||
|
||||
核心断言：
|
||||
- `_gilbert_positions` 与参考实现（逐点 append 版 gilbert2d，原样抄录）逐元素
|
||||
一致 —— 这是「与网页版互解」的关键（曲线必须同一个）；
|
||||
- 位置序列恒为 0..total-1 的排列；
|
||||
- encrypt / decrypt 互为逆操作（数组级 + PIL 级），偏移公式与参考一致。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib.util
|
||||
import io
|
||||
import math
|
||||
import random
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import numpy as np
|
||||
import pytest
|
||||
from PIL import Image
|
||||
|
||||
_MODULE_PATH = (
|
||||
Path(__file__).resolve().parents[1]
|
||||
/ "hexi"
|
||||
/ "plugins"
|
||||
/ "nonebot_plugin_cherry_tomato"
|
||||
/ "utils"
|
||||
/ "obfuscation.py"
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def obf():
|
||||
"""以独立模块名加载 obfuscation.py，避免触发插件包 __init__ 的依赖链。"""
|
||||
spec = importlib.util.spec_from_file_location(
|
||||
"hexi_cherry_tomato_obfuscation_test", _MODULE_PATH
|
||||
)
|
||||
assert spec is not None and spec.loader is not None
|
||||
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)
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────────────────────
|
||||
# 参考实现（Cherry-tomato-image-obfuscation 逐点 append 版），
|
||||
# 原样抄录，仅作对照，不执行于生产路径
|
||||
# ─────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def _naive_gilbert2d(width: int, height: int) -> list[tuple[int, int]]:
|
||||
coordinates: list[tuple[int, int]] = []
|
||||
if width >= height:
|
||||
_naive_generate2d(0, 0, width, 0, 0, height, coordinates)
|
||||
else:
|
||||
_naive_generate2d(0, 0, 0, height, width, 0, coordinates)
|
||||
return coordinates
|
||||
|
||||
|
||||
def _naive_sign(x):
|
||||
if x > 0:
|
||||
return 1
|
||||
elif x < 0:
|
||||
return -1
|
||||
return 0
|
||||
|
||||
|
||||
def _naive_generate2d(x, y, ax, ay, bx, by, coordinates):
|
||||
w = abs(ax + ay)
|
||||
h = abs(bx + by)
|
||||
|
||||
dax = _naive_sign(ax)
|
||||
day = _naive_sign(ay)
|
||||
dbx = _naive_sign(bx)
|
||||
dby = _naive_sign(by)
|
||||
|
||||
if h == 1:
|
||||
for _ in range(w):
|
||||
coordinates.append((x, y))
|
||||
x += dax
|
||||
y += day
|
||||
return
|
||||
|
||||
if w == 1:
|
||||
for _ in range(h):
|
||||
coordinates.append((x, y))
|
||||
x += dbx
|
||||
y += dby
|
||||
return
|
||||
|
||||
ax2 = ax // 2
|
||||
ay2 = ay // 2
|
||||
bx2 = bx // 2
|
||||
by2 = by // 2
|
||||
|
||||
w2 = abs(ax2 + ay2)
|
||||
h2 = abs(bx2 + by2)
|
||||
|
||||
if 2 * w > 3 * h:
|
||||
if (w2 % 2) and (w > 2):
|
||||
ax2 += dax
|
||||
ay2 += day
|
||||
_naive_generate2d(x, y, ax2, ay2, bx, by, coordinates)
|
||||
_naive_generate2d(x + ax2, y + ay2, ax - ax2, ay - ay2, bx, by, coordinates)
|
||||
else:
|
||||
if (h2 % 2) and (h > 2):
|
||||
bx2 += dbx
|
||||
by2 += dby
|
||||
_naive_generate2d(x, y, bx2, by2, ax2, ay2, coordinates)
|
||||
_naive_generate2d(x + bx2, y + by2, ax, ay, bx - bx2, by - by2, coordinates)
|
||||
_naive_generate2d(
|
||||
x + (ax - dax) + (bx2 - dbx),
|
||||
y + (ay - day) + (by2 - dby),
|
||||
-bx2,
|
||||
-by2,
|
||||
-(ax - ax2),
|
||||
-(ay - ay2),
|
||||
coordinates,
|
||||
)
|
||||
|
||||
|
||||
def _naive_positions(width: int, height: int) -> list[int]:
|
||||
return [x + y * width for x, y in _naive_gilbert2d(width, height)]
|
||||
|
||||
|
||||
def _build_sizes() -> list[tuple[int, int]]:
|
||||
sizes: list[tuple[int, int]] = [
|
||||
(1, 1),
|
||||
(1, 2),
|
||||
(2, 1),
|
||||
(1, 3),
|
||||
(3, 1),
|
||||
(2, 2),
|
||||
(3, 3),
|
||||
(7, 5),
|
||||
(5, 7),
|
||||
(1, 33),
|
||||
(33, 1),
|
||||
(1, 400),
|
||||
(400, 1),
|
||||
(128, 96),
|
||||
(200, 50),
|
||||
(50, 200),
|
||||
(333, 37),
|
||||
(256, 256),
|
||||
]
|
||||
primes = (2, 3, 5, 13, 31, 61, 79)
|
||||
sizes += [(p, q) for p in primes for q in primes]
|
||||
rng = random.Random(20261010)
|
||||
sizes += [(rng.randint(1, 80), rng.randint(1, 80)) for _ in range(220)]
|
||||
seen: set[tuple[int, int]] = set()
|
||||
unique: list[tuple[int, int]] = []
|
||||
for size in sizes:
|
||||
if size not in seen:
|
||||
seen.add(size)
|
||||
unique.append(size)
|
||||
return unique
|
||||
|
||||
|
||||
_SIZES = _build_sizes()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("size", _SIZES, ids=[f"{w}x{h}" for w, h in _SIZES])
|
||||
def test_positions_match_reference(obf, size):
|
||||
"""任何尺寸下位置序列都与参考逐点实现完全一致（互解的前提）。"""
|
||||
width, height = size
|
||||
pos = obf._gilbert_positions(width, height)
|
||||
assert pos.dtype == np.int32
|
||||
assert pos.tolist() == _naive_positions(width, height)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("size", _SIZES, ids=[f"{w}x{h}" for w, h in _SIZES])
|
||||
def test_positions_is_permutation(obf, size):
|
||||
"""位置序列必须是 0..total-1 的排列（不重不漏）。"""
|
||||
width, height = size
|
||||
pos = obf._gilbert_positions(width, height)
|
||||
assert sorted(pos.tolist()) == list(range(width * height))
|
||||
|
||||
|
||||
def test_gilbert_offset_formula(obf):
|
||||
"""黄金比例偏移与参考/JS 公式一致。"""
|
||||
for total in (1, 2, 3, 6, 10, 1920 * 1080, 8_000_000):
|
||||
assert obf.gilbert_offset(total) == round((math.sqrt(5) - 1) / 2 * total)
|
||||
assert obf.gilbert_offset(10) == 6 # 0.618...*10 = 6.18 → 6（与 JS 一致）
|
||||
|
||||
|
||||
@pytest.mark.parametrize("size", [(3, 2), (17, 5), (40, 33), (64, 64)])
|
||||
def test_encrypt_decrypt_array_roundtrip(obf, size):
|
||||
"""数组级 encrypt∘decrypt 恒等；encrypt 不改输入、确实重排了像素。"""
|
||||
width, height = size
|
||||
rng = np.random.default_rng(20261010)
|
||||
src = rng.integers(0, 256, size=(width * height, 3), dtype=np.uint8)
|
||||
original = src.copy()
|
||||
|
||||
enc = obf.encrypt_pixels(src, width, height)
|
||||
assert np.array_equal(src, original) # 输入未被改写
|
||||
assert not np.array_equal(enc, src) # 偏移非 0，随机图必被重排
|
||||
|
||||
dec = obf.decrypt_pixels(enc, width, height)
|
||||
assert np.array_equal(dec, src)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("size", [(3, 2), (13, 9), (40, 33)])
|
||||
def test_encrypt_decrypt_pil_roundtrip(obf, size):
|
||||
"""PIL 级 round-trip：obfuscate 输出为 RGB、无 EXIF，解混淆还原原图。"""
|
||||
width, height = size
|
||||
rng = np.random.default_rng(7)
|
||||
arr = rng.integers(0, 256, (height, width, 3), dtype=np.uint8)
|
||||
img = Image.fromarray(arr, "RGB")
|
||||
|
||||
enc, scaled, extra = obf.obfuscate(img, decrypt=False, max_pixels=10**9)
|
||||
assert enc.mode == "RGB"
|
||||
assert scaled is False
|
||||
assert extra is False
|
||||
assert not np.array_equal(np.asarray(enc), arr)
|
||||
|
||||
dec, _, _ = obf.obfuscate(enc, decrypt=True, max_pixels=10**9)
|
||||
assert np.array_equal(np.asarray(dec), arr)
|
||||
assert not dec.getexif() # 输出统一不带 EXIF
|
||||
|
||||
# JPEG 编码可解码、尺寸不变
|
||||
data = obf.to_jpeg_bytes(dec, 95)
|
||||
with Image.open(io.BytesIO(data)) as reopened:
|
||||
assert reopened.size == (width, height)
|
||||
|
||||
|
||||
def test_obfuscate_applies_exif_orientation(obf):
|
||||
"""带 EXIF 旋转的照片先规范化方向：输出宽高对调（显示方向不突变）。"""
|
||||
img = Image.new("RGB", (13, 9), (10, 20, 30))
|
||||
exif = Image.Exif()
|
||||
exif[0x0112] = 6 # Orientation=6：显示时需顺时针旋转 90°（宽高对调）
|
||||
buf = io.BytesIO()
|
||||
img.save(buf, format="JPEG", exif=exif)
|
||||
buf.seek(0)
|
||||
with Image.open(buf) as photo:
|
||||
assert photo.size == (13, 9)
|
||||
assert photo.getexif().get(0x0112) == 6
|
||||
out, _, _ = obf.obfuscate(photo, decrypt=False, max_pixels=10**9)
|
||||
assert out.size == (9, 13)
|
||||
|
||||
|
||||
def test_fit_within(obf):
|
||||
"""fit_within：未超限原样返回；超限等比缩小且至少 1 像素。"""
|
||||
img = Image.new("RGB", (4000, 3000))
|
||||
assert obf.fit_within(img, 12_000_000) is img
|
||||
|
||||
out = obf.fit_within(img, 2_000_000)
|
||||
assert out.width * out.height <= 2_000_000
|
||||
assert abs(out.width / out.height - 4000 / 3000) < 0.01 # 等比
|
||||
|
||||
tiny = obf.fit_within(img, 1)
|
||||
assert tiny.width >= 1 and tiny.height >= 1
|
||||
|
||||
|
||||
def test_obfuscate_scale_flag_only_for_encrypt(obf):
|
||||
"""加密超限缩放并置 scaled；解密绝不缩放（超大由 pipeline 拒绝兜底）。"""
|
||||
img = Image.new("RGB", (60, 40))
|
||||
|
||||
enc, scaled, _ = obf.obfuscate(img, decrypt=False, max_pixels=1000)
|
||||
assert scaled is True
|
||||
assert enc.width * enc.height <= 1000
|
||||
|
||||
dec, scaled2, _ = obf.obfuscate(img, decrypt=True, max_pixels=1000)
|
||||
assert scaled2 is False
|
||||
assert dec.size == img.size
|
||||
|
||||
|
||||
def test_obfuscate_extra_frames_flag(obf):
|
||||
"""多帧图只取第一帧并置 extra_frames；单帧图不置。"""
|
||||
frames = [
|
||||
Image.new("RGB", (8, 6), (255, 0, 0)),
|
||||
Image.new("RGB", (8, 6), (0, 0, 255)),
|
||||
]
|
||||
buf = io.BytesIO()
|
||||
frames[0].save(buf, format="GIF", save_all=True, append_images=frames[1:])
|
||||
buf.seek(0)
|
||||
with Image.open(buf) as gif:
|
||||
out, _, extra = obf.obfuscate(gif, decrypt=False, max_pixels=10**9)
|
||||
assert extra is True
|
||||
assert out.size == (8, 6)
|
||||
# 首帧是纯红：均匀图重排后仍是纯红（同时证明没拿第二帧的蓝色）
|
||||
assert out.getpixel((0, 0)) == (255, 0, 0)
|
||||
|
||||
_, _, extra2 = obf.obfuscate(
|
||||
Image.new("RGB", (8, 6)), decrypt=False, max_pixels=10**9
|
||||
)
|
||||
assert extra2 is False
|
||||
Reference in new issue
Block a user