forked from mttu-developers/konabot
Merge pull request 'feat: ytpgif' (#16) from tnot/konabot:feat--ytpgif into master
Reviewed-on: mttu-developers/konabot#16
This commit is contained in:
212
konabot/plugins/ytpgif/__init__.py
Normal file
212
konabot/plugins/ytpgif/__init__.py
Normal file
@ -0,0 +1,212 @@
|
||||
import os
|
||||
import tempfile
|
||||
from typing import Optional
|
||||
|
||||
from PIL import Image, ImageSequence
|
||||
from nonebot.adapters import Event as BaseEvent
|
||||
from nonebot.plugin import PluginMetadata
|
||||
from nonebot_plugin_alconna import (
|
||||
Alconna,
|
||||
Args,
|
||||
Field,
|
||||
UniMessage,
|
||||
on_alconna,
|
||||
)
|
||||
|
||||
__plugin_meta__ = PluginMetadata(
|
||||
name="ytpgif",
|
||||
description="生成来回镜像翻转的动图：动图按时间分段播放，静态图高频翻转（带参数范围保护）",
|
||||
usage="/ytpgif [倍速=1.0] （倍速范围：0.1～20.0）",
|
||||
type="application",
|
||||
config=None,
|
||||
homepage=None,
|
||||
)
|
||||
|
||||
# 参数定义（带硬性限制）
|
||||
BASE_SEGMENT_DURATION = 0.25
|
||||
BASE_INTERVAL = 0.25
|
||||
MAX_SIZE = 256
|
||||
MIN_SPEED = 0.1
|
||||
MAX_SPEED = 20.0
|
||||
MAX_FRAMES_PER_SEGMENT = 500
|
||||
|
||||
# 提示语
|
||||
SPEED_TIPS = f"倍速必须是 {MIN_SPEED} 到 {MAX_SPEED} 之间的数字"
|
||||
|
||||
|
||||
# 定义命令 + 参数校验
|
||||
ytpgif_cmd = on_alconna(
|
||||
Alconna(
|
||||
"ytpgif",
|
||||
Args[
|
||||
"speed?",
|
||||
float,
|
||||
Field(
|
||||
default=1.0,
|
||||
unmatch_tips=lambda x: f"“{x}”不是有效数值。{SPEED_TIPS}",
|
||||
),
|
||||
],
|
||||
),
|
||||
use_cmd_start=True,
|
||||
use_cmd_sep=False,
|
||||
skip_for_unmatch=False,
|
||||
)
|
||||
|
||||
|
||||
async def get_image_url(event: BaseEvent) -> Optional[str]:
|
||||
msg = event.get_message()
|
||||
for seg in msg:
|
||||
if seg.type == "image" and seg.data.get("url"):
|
||||
return str(seg.data["url"])
|
||||
if hasattr(event, "reply") and (reply := event.reply):
|
||||
for seg in reply.message:
|
||||
if seg.type == "image" and seg.data.get("url"):
|
||||
return str(seg.data["url"])
|
||||
return None
|
||||
|
||||
|
||||
async def download_image(url: str) -> bytes:
|
||||
import httpx
|
||||
async with httpx.AsyncClient() as client:
|
||||
resp = await client.get(url, timeout=10)
|
||||
resp.raise_for_status()
|
||||
return resp.content
|
||||
|
||||
|
||||
def resize_frame(frame: Image.Image) -> Image.Image:
|
||||
w, h = frame.size
|
||||
if w <= MAX_SIZE and h <= MAX_SIZE:
|
||||
return frame
|
||||
scale = MAX_SIZE / max(w, h)
|
||||
new_w = int(w * scale)
|
||||
new_h = int(h * scale)
|
||||
return frame.resize((new_w, new_h), Image.Resampling.LANCZOS)
|
||||
|
||||
|
||||
@ytpgif_cmd.handle()
|
||||
async def handle_ytpgif(event: BaseEvent, speed: float = 1.0):
|
||||
# === 校验 speed 范围 ===
|
||||
if not (MIN_SPEED <= speed <= MAX_SPEED):
|
||||
await ytpgif_cmd.send(
|
||||
await UniMessage.text(f"❌ {SPEED_TIPS}").export()
|
||||
)
|
||||
return
|
||||
|
||||
img_url = await get_image_url(event)
|
||||
if not img_url:
|
||||
await ytpgif_cmd.send(
|
||||
await UniMessage.text(
|
||||
"请发送一张图片并使用 /ytpgif，或回复一张图片来生成镜像动图。"
|
||||
).export()
|
||||
)
|
||||
return
|
||||
|
||||
try:
|
||||
image_data = await download_image(img_url)
|
||||
except Exception:
|
||||
await ytpgif_cmd.send(
|
||||
await UniMessage.text("❌ 图片下载失败，请重试。").export()
|
||||
)
|
||||
return
|
||||
|
||||
input_path = output_path = None
|
||||
try:
|
||||
with tempfile.NamedTemporaryFile(delete=False, suffix=".gif") as tmp_in:
|
||||
tmp_in.write(image_data)
|
||||
input_path = tmp_in.name
|
||||
|
||||
with tempfile.NamedTemporaryFile(delete=False, suffix=".gif") as tmp_out:
|
||||
output_path = tmp_out.name
|
||||
|
||||
with Image.open(input_path) as src_img:
|
||||
is_animated = getattr(src_img, "is_animated", False) or src_img.n_frames > 1
|
||||
|
||||
output_frames = []
|
||||
output_durations_ms = []
|
||||
|
||||
if is_animated:
|
||||
# === 动图模式：播放两段，每段最多 BASE_SEGMENT_DURATION * speed 秒，且帧数 ≤ 100 ===
|
||||
frames_with_duration = []
|
||||
for frame in ImageSequence.Iterator(src_img):
|
||||
rgb_frame = frame.convert("RGB")
|
||||
resized_frame = resize_frame(rgb_frame)
|
||||
ms = frame.info.get("duration", int(BASE_SEGMENT_DURATION * 1000))
|
||||
dur_sec = max(0.01, ms / 1000.0) # 至少 10ms
|
||||
frames_with_duration.append((resized_frame, dur_sec))
|
||||
|
||||
max_dur = BASE_SEGMENT_DURATION * speed # 每段最大播放时间
|
||||
accumulated = 0.0
|
||||
frame_count = 0
|
||||
|
||||
# 正向段
|
||||
for img, dur in frames_with_duration:
|
||||
if accumulated + dur > max_dur or frame_count >= MAX_FRAMES_PER_SEGMENT:
|
||||
break
|
||||
output_frames.append(img)
|
||||
output_durations_ms.append(int(dur * 1000))
|
||||
accumulated += dur
|
||||
frame_count += 1
|
||||
|
||||
if frame_count == 0:
|
||||
await ytpgif_cmd.send(
|
||||
await UniMessage.text("动图帧太短，无法生成有效片段。").export()
|
||||
)
|
||||
return
|
||||
|
||||
# 镜像段（从头开始）
|
||||
accumulated = 0.0
|
||||
frame_count = 0
|
||||
for img, dur in frames_with_duration:
|
||||
if accumulated + dur > max_dur or frame_count >= MAX_FRAMES_PER_SEGMENT:
|
||||
break
|
||||
flipped = img.transpose(Image.FLIP_LEFT_RIGHT)
|
||||
output_frames.append(flipped)
|
||||
output_durations_ms.append(int(dur * 1000))
|
||||
accumulated += dur
|
||||
frame_count += 1
|
||||
|
||||
else:
|
||||
# === 静态图模式：制作翻转动画 ===
|
||||
raw_frame = src_img.convert("RGB")
|
||||
resized_frame = resize_frame(raw_frame)
|
||||
|
||||
interval_sec = max(0.025, min(2.5, BASE_INTERVAL / speed)) # 限制在 25ms ~ 2.5s
|
||||
duration_ms = int(interval_sec * 1000)
|
||||
|
||||
frame1 = resized_frame
|
||||
frame2 = resized_frame.transpose(Image.FLIP_LEFT_RIGHT)
|
||||
|
||||
output_frames = [frame1, frame2]
|
||||
output_durations_ms = [duration_ms, duration_ms]
|
||||
|
||||
if len(output_frames) < 1:
|
||||
await ytpgif_cmd.send(
|
||||
await UniMessage.text("未能生成任何帧。").export()
|
||||
)
|
||||
return
|
||||
|
||||
# 保存 GIF
|
||||
output_frames[0].save(
|
||||
output_path,
|
||||
save_all=True,
|
||||
append_images=output_frames[1:],
|
||||
format="GIF",
|
||||
loop=0,
|
||||
duration=output_durations_ms,
|
||||
disposal=2,
|
||||
)
|
||||
|
||||
# 发送结果
|
||||
with open(output_path, "rb") as f:
|
||||
result_image = UniMessage.image(raw=f.read())
|
||||
await ytpgif_cmd.send(await result_image.export())
|
||||
|
||||
except Exception as e:
|
||||
print(f"[YTPGIF] 处理失败: {e}")
|
||||
await ytpgif_cmd.send(
|
||||
await UniMessage.text("❌ 处理失败，可能是图片格式不支持或文件过大。").export()
|
||||
)
|
||||
finally:
|
||||
for path in filter(None, [input_path, output_path]):
|
||||
if os.path.exists(path):
|
||||
os.unlink(path)
|
||||
Reference in New Issue
Block a user