sansenhoshiandClaude Code
fb15b976aa
Gilbert 曲线像素重排，可与网页版互解；`混淆` / `解混淆` 命令 （回复图片或图文同条消息发送），附 tests/test_cherry_tomato.py Co-Authored-By: Claude Code <noreply@anthropic.com>
70 lines
2.4 KiB
Python
70 lines
2.4 KiB
Python
"""小番茄图片混淆 —— 编排（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
|