diff --git a/hexi/plugins/nonebot_plugin_cherry_tomato/__init__.py b/hexi/plugins/nonebot_plugin_cherry_tomato/__init__.py new file mode 100644 index 0000000..e32e58e --- /dev/null +++ b/hexi/plugins/nonebot_plugin_cherry_tomato/__init__.py @@ -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() diff --git a/hexi/plugins/nonebot_plugin_cherry_tomato/config.py b/hexi/plugins/nonebot_plugin_cherry_tomato/config.py new file mode 100644 index 0000000..b9d8291 --- /dev/null +++ b/hexi/plugins/nonebot_plugin_cherry_tomato/config.py @@ -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__], + ) diff --git a/hexi/plugins/nonebot_plugin_cherry_tomato/handlers/__init__.py b/hexi/plugins/nonebot_plugin_cherry_tomato/handlers/__init__.py new file mode 100644 index 0000000..25c21da --- /dev/null +++ b/hexi/plugins/nonebot_plugin_cherry_tomato/handlers/__init__.py @@ -0,0 +1,3 @@ +"""小番茄图片混淆 —— Trigger 层。""" + +from .obfuscate import deobf_cmd, obf_cmd # noqa: F401 diff --git a/hexi/plugins/nonebot_plugin_cherry_tomato/handlers/obfuscate.py b/hexi/plugins/nonebot_plugin_cherry_tomato/handlers/obfuscate.py new file mode 100644 index 0000000..8a8bdae --- /dev/null +++ b/hexi/plugins/nonebot_plugin_cherry_tomato/handlers/obfuscate.py @@ -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() diff --git a/hexi/plugins/nonebot_plugin_cherry_tomato/services/__init__.py b/hexi/plugins/nonebot_plugin_cherry_tomato/services/__init__.py new file mode 100644 index 0000000..b184969 --- /dev/null +++ b/hexi/plugins/nonebot_plugin_cherry_tomato/services/__init__.py @@ -0,0 +1,3 @@ +"""小番茄图片混淆 —— Service 层。""" + +from .pipeline import download_image, run_pipeline # noqa: F401 diff --git a/hexi/plugins/nonebot_plugin_cherry_tomato/services/pipeline.py b/hexi/plugins/nonebot_plugin_cherry_tomato/services/pipeline.py new file mode 100644 index 0000000..1a3e086 --- /dev/null +++ b/hexi/plugins/nonebot_plugin_cherry_tomato/services/pipeline.py @@ -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 diff --git a/hexi/plugins/nonebot_plugin_cherry_tomato/utils/__init__.py b/hexi/plugins/nonebot_plugin_cherry_tomato/utils/__init__.py new file mode 100644 index 0000000..517551a --- /dev/null +++ b/hexi/plugins/nonebot_plugin_cherry_tomato/utils/__init__.py @@ -0,0 +1,10 @@ +"""小番茄图片混淆 —— 纯工具层(无 nonebot 依赖,可单测)。""" + +from .obfuscation import ( # noqa: F401 + decrypt_pixels, + encrypt_pixels, + fit_within, + gilbert_offset, + obfuscate, + to_jpeg_bytes, +) diff --git a/hexi/plugins/nonebot_plugin_cherry_tomato/utils/obfuscation.py b/hexi/plugins/nonebot_plugin_cherry_tomato/utils/obfuscation.py new file mode 100644 index 0000000..d6ad4fd --- /dev/null +++ b/hexi/plugins/nonebot_plugin_cherry_tomato/utils/obfuscation.py @@ -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() diff --git a/tests/test_cherry_tomato.py b/tests/test_cherry_tomato.py new file mode 100644 index 0000000..8fca24e --- /dev/null +++ b/tests/test_cherry_tomato.py @@ -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