Merge pull request 'feature-更多更多说' (#22) from feature-更多更多说 into master

Reviewed-on: #22

konabot/common/nb/extract_image.py

@@ -0,0 +1,135 @@
+from io import BytesIO
+
+import httpx
+import PIL.Image
+from loguru import logger
+from nonebot.adapters import Bot, Event, Message
+from nonebot.adapters.discord import Bot as DiscordBot
+from nonebot.adapters.onebot.v11 import Bot as OnebotV11Bot
+from nonebot.adapters.onebot.v11 import Message as OnebotV11Message
+from nonebot.adapters.onebot.v11 import MessageEvent as OnebotV11MessageEvent
+from nonebot_plugin_alconna import Image, RefNode, Reply, UniMessage
+from PIL import UnidentifiedImageError
+from returns.result import Failure, Result, Success
+
+
+async def download_image_bytes(url: str) -> Result[bytes, str]:
+    # if "/matcha/cache/" in url:
+    #     url = url.replace('127.0.0.1', '10.126.126.101')
+    logger.debug(f"开始从 {url} 下载图片")
+    async with httpx.AsyncClient() as c:
+        try:
+            response = await c.get(url)
+        except (httpx.ConnectError, httpx.RemoteProtocolError) as e:
+            return Failure(f"HTTPX 模块下载图片时出错：{e}")
+        except httpx.ConnectTimeout:
+            return Failure("下载图片失败了，网络超时了qwq")
+        if response.status_code != 200:
+            return Failure("无法下载图片，可能存在网络问题需要排查")
+        return Success(response.content)
+
+
+def bytes_to_pil(raw_data: bytes | BytesIO) -> Result[PIL.Image.Image, str]:
+    try:
+        if not isinstance(raw_data, BytesIO):
+            img_pil = PIL.Image.open(BytesIO(raw_data))
+        else:
+            img_pil = PIL.Image.open(raw_data)
+        img_pil.verify()
+        if not isinstance(raw_data, BytesIO):
+            img = PIL.Image.open(BytesIO(raw_data))
+        else:
+            raw_data.seek(0)
+            img = PIL.Image.open(raw_data)
+        return Success(img)
+    except UnidentifiedImageError:
+        return Failure("图像无法读取，可能是格式不支持orz")
+    except IOError:
+        return Failure("图像无法读取，可能是网络存在问题orz")
+
+
+async def unimsg_img_to_pil(image: Image) -> Result[PIL.Image.Image, str]:
+    if image.url is not None:
+        raw_result = await download_image_bytes(image.url)
+    elif image.raw is not None:
+        raw_result = Success(image.raw)
+    else:
+        return Failure("由于一些内部问题，下载图片失败了orz")
+
+    return raw_result.bind(bytes_to_pil)
+
+
+async def extract_image_from_qq_message(
+    msg: OnebotV11Message,
+    evt: OnebotV11MessageEvent,
+    bot: OnebotV11Bot,
+    allow_reply: bool = True,
+) -> Result[PIL.Image.Image, str]:
+    if allow_reply and (reply := evt.reply) is not None:
+        return await extract_image_from_qq_message(
+            reply.message,
+            evt,
+            bot,
+            False,
+        )
+    for seg in msg:
+        if seg.type == "reply" and allow_reply:
+            msgid = seg.data.get("id")
+            if msgid is None:
+                return Failure("消息可能太久远，无法读取到消息原文")
+            try:
+                msg2 = await bot.get_msg(message_id=msgid)
+            except Exception as e:
+                logger.warning(f"获取消息内容时出错：{e}")
+                return Failure("消息可能太久远，无法读取到消息原文")
+            msg2_data = msg2.get("message")
+            if msg2_data is None:
+                return Failure("消息可能太久远，无法读取到消息原文")
+            logger.debug("发现消息引用，递归一层")
+            return await extract_image_from_qq_message(
+                msg=OnebotV11Message(msg2_data),
+                evt=evt,
+                bot=bot,
+                allow_reply=False,
+            )
+        if seg.type == "image":
+            url = seg.data.get("url")
+            if url is None:
+                return Failure("无法下载图片，可能有一些网络问题")
+            data = await download_image_bytes(url)
+            return data.bind(bytes_to_pil)
+
+    return Failure("请在消息中包含图片，或者引用一个含有图片的消息")
+
+
+async def extract_image_from_message(
+    msg: Message,
+    evt: Event,
+    bot: Bot,
+    allow_reply: bool = True,
+) -> Result[PIL.Image.Image, str]:
+    if (
+        isinstance(bot, OnebotV11Bot) 
+        and isinstance(msg, OnebotV11Message) 
+        and isinstance(evt, OnebotV11MessageEvent)
+    ):
+        # 看起来 UniMessage 在这方面能力似乎不足，因此用 QQ 的
+        logger.debug('获取图片的路径 Fallback 到 QQ 模块')
+        return await extract_image_from_qq_message(msg, evt, bot, allow_reply)
+
+    for seg in UniMessage.of(msg, bot):
+        logger.info(seg)
+        if isinstance(seg, Image):
+            return await unimsg_img_to_pil(seg)
+        elif isinstance(seg, Reply) and allow_reply:
+            msg2 = seg.msg
+            logger.debug(f"深入搜索引用的消息：{msg2}")
+            if msg2 is None or isinstance(msg2, str):
+                continue
+            return await extract_image_from_message(msg2, evt, bot, False)
+        elif isinstance(seg, RefNode) and allow_reply:
+            if isinstance(bot, DiscordBot):
+                return Failure("暂时不支持在 Discord 中通过引用的方式获取图片")
+            else:
+                return Failure("暂时不支持在这里中通过引用的方式获取图片")
+    return Failure("请在消息中包含图片，或者引用一个含有图片的消息")

konabot/plugins/memepack/__init__.py

@@ -1,10 +1,19 @@
 from io import BytesIO
+from typing import Iterable, cast
 
-from nonebot_plugin_alconna import (Alconna, Args, Field, MultiVar, UniMessage,
-                                    on_alconna)
+from nonebot import on_message
+from nonebot_plugin_alconna import (Alconna, Args, Field, MultiVar, Text,
+                                    UniMessage, UniMsg, on_alconna)
 
-from konabot.plugins.memepack.drawing.geimao import draw_geimao
-from konabot.plugins.memepack.drawing.pt import draw_pt
+from konabot.common.nb.extract_image import extract_image_from_message
+from konabot.plugins.memepack.drawing.display import draw_cao_display
+from konabot.plugins.memepack.drawing.saying import (draw_cute_ten,
+                                                     draw_geimao, draw_mnk,
+                                                     draw_pt, draw_suan)
+
+from nonebot.adapters import Bot, Event
+
+from returns.result import Success, Failure
 
 geimao = on_alconna(Alconna(
     "给猫说",
@@ -36,3 +45,97 @@ async def _(saying: list[str]):
     img.save(img_bytes, format="PNG")
 
     await pt.send(await UniMessage().image(raw=img_bytes).export())
+
+
+mnk = on_alconna(Alconna(
+    "re:小?黑白子?说",
+    Args["saying", MultiVar(str, '+'), Field(
+        missing_tips=lambda: "你没有写黑白子说了什么"
+    )]
+), use_cmd_start=True, use_cmd_sep=False, skip_for_unmatch=False, aliases={"mnk说"})
+
+@mnk.handle()
+async def _(saying: list[str]):
+    img = await draw_mnk("\n".join(saying))
+    img_bytes = BytesIO()
+    img.save(img_bytes, format="PNG")
+
+    await mnk.send(await UniMessage().image(raw=img_bytes).export())
+
+
+suan = on_alconna(Alconna(
+    "小蒜说",
+    Args["saying", MultiVar(str, '+'), Field(
+        missing_tips=lambda: "你没有写小蒜说了什么"
+    )]
+), use_cmd_start=True, use_cmd_sep=False, skip_for_unmatch=False, aliases=set())
+
+@suan.handle()
+async def _(saying: list[str]):
+    img = await draw_suan("\n".join(saying))
+    img_bytes = BytesIO()
+    img.save(img_bytes, format="PNG")
+
+    await suan.send(await UniMessage().image(raw=img_bytes).export())
+
+
+dsuan = on_alconna(Alconna(
+    "大蒜说",
+    Args["saying", MultiVar(str, '+'), Field(
+        missing_tips=lambda: "你没有写大蒜说了什么"
+    )]
+), use_cmd_start=True, use_cmd_sep=False, skip_for_unmatch=False, aliases=set())
+
+@dsuan.handle()
+async def _(saying: list[str]):
+    img = await draw_suan("\n".join(saying), True)
+    img_bytes = BytesIO()
+    img.save(img_bytes, format="PNG")
+
+    await dsuan.send(await UniMessage().image(raw=img_bytes).export())
+
+
+cutecat = on_alconna(Alconna(
+    "乖猫说",
+    Args["saying", MultiVar(str, '+'), Field(
+        missing_tips=lambda: "你没有写十猫说了什么"
+    )]
+), use_cmd_start=True, use_cmd_sep=False, skip_for_unmatch=False, aliases={"十猫说"})
+
+@cutecat.handle()
+async def _(saying: list[str]):
+    img = await draw_cute_ten("\n".join(saying))
+    img_bytes = BytesIO()
+    img.save(img_bytes, format="PNG")
+
+    await cutecat.send(await UniMessage().image(raw=img_bytes).export())
+
+
+cao_display_cmd = on_message()
+
+@cao_display_cmd.handle()
+async def _(msg: UniMsg, evt: Event, bot: Bot):
+    flag = False
+    for text in cast(Iterable[Text], msg.get(Text)):
+        if text.text.strip() == "小槽展示":
+            flag = True
+        elif text.text.strip() == '':
+            continue
+        else:
+            return
+    if not flag:
+        return
+    match await extract_image_from_message(evt.get_message(), evt, bot):
+        case Success(img):
+            img_handled = await draw_cao_display(img)
+            img_bytes = BytesIO()
+            img_handled.save(img_bytes, format="PNG")
+            await cao_display_cmd.send(await UniMessage().image(raw=img_bytes).export())
+        case Failure(err):
+            await cao_display_cmd.send(
+                await UniMessage()
+                    .at(user_id=evt.get_user_id())
+                    .text(' ')
+                    .text(err)
+                    .export()
+            )

konabot/plugins/memepack/drawing/base/fonts.py

@@ -10,3 +10,4 @@ FontDB.SetDefaultEmojiOptions(EmojiOptions(
 
 HARMONYOS_SANS_SC_BLACK = FontDB.Query("HarmonyOS_Sans_SC_Black")
 HARMONYOS_SANS_SC_REGULAR = FontDB.Query("HarmonyOS_Sans_SC_Regular")
+LXGWWENKAI_REGULAR = FontDB.Query("LXGWWenKai-Regular")

konabot/plugins/memepack/drawing/display.py

@@ -0,0 +1,45 @@
+import asyncio
+from typing import Any, cast
+
+import cv2
+import numpy as np
+import PIL.Image
+
+from konabot.common.path import ASSETS_PATH
+
+cao_image = PIL.Image.open(ASSETS_PATH / "img" / "meme" / "caoimg1.png")
+CAO_QUAD_POINTS = np.float32(cast(Any, [
+    [392, 540],
+    [577, 557],
+    [567, 707],
+    [381, 687],
+]))
+
+def _draw_cao_display(image: PIL.Image.Image):
+    src = np.array(image.convert("RGB"))
+    h, w = src.shape[:2]
+    src_points = np.float32(cast(Any, [
+        [0, 0],
+        [w, 0],
+        [w, h],
+        [0, h]
+    ]))
+    dst_points = CAO_QUAD_POINTS
+    M = cv2.getPerspectiveTransform(cast(Any, src_points), cast(Any, dst_points))
+    output_size = cao_image.size
+    output_w, output_h = output_size
+    warped = cv2.warpPerspective(
+        src,
+        M,
+        (output_w, output_h),
+        flags=cv2.INTER_LINEAR,
+        borderMode=cv2.BORDER_CONSTANT,
+        borderValue=(0, 0, 0)
+    )
+    result = PIL.Image.fromarray(warped, 'RGB').convert('RGBA')
+    result = PIL.Image.alpha_composite(result, cao_image)
+    return result
+
+
+async def draw_cao_display(image: PIL.Image.Image):
+    return await asyncio.to_thread(_draw_cao_display, image)

konabot/plugins/memepack/drawing/geimao.py

@@ -1,30 +0,0 @@
-import asyncio
-from typing import Any, cast
-
-import imagetext_py
-import PIL.Image
-
-from konabot.common.path import ASSETS_PATH
-
-from .base.fonts import HARMONYOS_SANS_SC_BLACK
-
-geimao_image = PIL.Image.open(ASSETS_PATH / "img" / "meme" / "geimao.jpg").convert("RGBA")
-
-
-def _draw_geimao(saying: str):
-    img = geimao_image.copy()
-    with imagetext_py.Writer(img) as iw:
-        iw.draw_text_wrapped(
-            saying, 960, 50, 00.5, 0, 1920, 240, HARMONYOS_SANS_SC_BLACK,
-            imagetext_py.Paint.Color(imagetext_py.Color.from_hex("000000FF")),
-            0.8,
-            imagetext_py.TextAlign.Center,
-            cast(Any, 30.0),
-            imagetext_py.Paint.Color(imagetext_py.Color.from_hex("FFFFFFFF")),
-            draw_emojis=True,
-        )
-    return img
-    
-
-async def draw_geimao(saying: str):
-    return await asyncio.to_thread(_draw_geimao, saying)

konabot/plugins/memepack/drawing/pt.py

@@ -1,27 +0,0 @@
-import asyncio
-
-import imagetext_py
-import PIL.Image
-
-from konabot.common.path import ASSETS_PATH
-
-from .base.fonts import HARMONYOS_SANS_SC_REGULAR
-
-pt_image = PIL.Image.open(ASSETS_PATH / "img" / "meme" / "ptsay.png").convert("RGBA")
-
-
-def _draw_pt(saying: str):
-    img = pt_image.copy()
-    with imagetext_py.Writer(img) as iw:
-        iw.draw_text_wrapped(
-            saying, 259, 278, 0.5, 0.5, 360, 48, HARMONYOS_SANS_SC_REGULAR,
-            imagetext_py.Paint.Color(imagetext_py.Color.from_hex("000000FF")),
-            1.0,
-            imagetext_py.TextAlign.Center,
-            draw_emojis=True,
-        )
-    return img
-
-
-async def draw_pt(saying: str):
-    return await asyncio.to_thread(_draw_pt, saying)

konabot/plugins/memepack/drawing/saying.py

@@ -0,0 +1,108 @@
+import asyncio
+from typing import Any, cast
+
+import imagetext_py
+import PIL.Image
+
+from konabot.common.path import ASSETS_PATH
+
+from .base.fonts import HARMONYOS_SANS_SC_BLACK, HARMONYOS_SANS_SC_REGULAR, LXGWWENKAI_REGULAR
+
+geimao_image = PIL.Image.open(ASSETS_PATH / "img" / "meme" / "geimao.jpg").convert("RGBA")
+pt_image = PIL.Image.open(ASSETS_PATH / "img" / "meme" / "ptsay.png").convert("RGBA")
+mnk_image = PIL.Image.open(ASSETS_PATH / "img" / "meme" / "mnksay.jpg").convert("RGBA")
+dasuan_image = PIL.Image.open(ASSETS_PATH / "img" / "meme" / "dss.png").convert("RGBA")
+suan_image = PIL.Image.open(ASSETS_PATH / "img" / "meme" / "suanleba.png").convert("RGBA")
+cute_ten_image = PIL.Image.open(ASSETS_PATH / "img" / "meme" / "tententen.png").convert("RGBA")
+
+
+def _draw_geimao(saying: str):
+    img = geimao_image.copy()
+    with imagetext_py.Writer(img) as iw:
+        iw.draw_text_wrapped(
+            saying, 960, 50, 0.5, 0, 1920, 240, HARMONYOS_SANS_SC_BLACK,
+            imagetext_py.Paint.Color(imagetext_py.Color.from_hex("000000FF")),
+            0.8,
+            imagetext_py.TextAlign.Center,
+            cast(Any, 30.0),
+            imagetext_py.Paint.Color(imagetext_py.Color.from_hex("FFFFFFFF")),
+            draw_emojis=True,
+        )
+    return img
+    
+
+async def draw_geimao(saying: str):
+    return await asyncio.to_thread(_draw_geimao, saying)
+
+
+def _draw_pt(saying: str):
+    img = pt_image.copy()
+    with imagetext_py.Writer(img) as iw:
+        iw.draw_text_wrapped(
+            saying, 259, 278, 0.5, 0.5, 360, 48, HARMONYOS_SANS_SC_REGULAR,
+            imagetext_py.Paint.Color(imagetext_py.Color.from_hex("000000FF")),
+            1.0,
+            imagetext_py.TextAlign.Center,
+            draw_emojis=True,
+        )
+    return img
+
+
+async def draw_pt(saying: str):
+    return await asyncio.to_thread(_draw_pt, saying)
+
+
+def _draw_mnk(saying: str):
+    img = mnk_image.copy()
+    with imagetext_py.Writer(img) as iw:
+        iw.draw_text_wrapped(
+            saying, 540, 25, 0.5, 0, 1080, 120, HARMONYOS_SANS_SC_BLACK,
+            imagetext_py.Paint.Color(imagetext_py.Color.from_hex("000000FF")),
+            0.8,
+            imagetext_py.TextAlign.Center,
+            cast(Any, 15.0),
+            imagetext_py.Paint.Color(imagetext_py.Color.from_hex("FFFFFFFF")),
+            draw_emojis=True,
+        )
+    return img
+
+
+async def draw_mnk(saying: str):
+    return await asyncio.to_thread(_draw_mnk, saying)
+
+
+def _draw_suan(saying: str, dasuan: bool = False):
+    if dasuan:
+        img = dasuan_image.copy()
+    else:
+        img = suan_image.copy()
+    with imagetext_py.Writer(img) as iw:
+        iw.draw_text_wrapped(
+            saying, 1020, 290, 0.5, 0.5, 400, 48, LXGWWENKAI_REGULAR,
+            imagetext_py.Paint.Color(imagetext_py.Color.from_hex("000000FF")),
+            1.0,
+            imagetext_py.TextAlign.Center,
+            draw_emojis=True,
+        )
+    return img
+
+
+async def draw_suan(saying: str, dasuan: bool = False):
+    return await asyncio.to_thread(_draw_suan, saying, dasuan)
+
+
+def _draw_cute_ten(saying: str):
+    img = cute_ten_image.copy()
+    with imagetext_py.Writer(img) as iw:
+        iw.draw_text_wrapped(
+            saying, 390, 479, 0.5, 0.5, 760, 96, LXGWWENKAI_REGULAR,
+            imagetext_py.Paint.Color(imagetext_py.Color.from_hex("000000FF")),
+            1.0,
+            imagetext_py.TextAlign.Center,
+            draw_emojis=True,
+        )
+    return img
+
+
+async def draw_cute_ten(saying: str):
+    return await asyncio.to_thread(_draw_cute_ten, saying)

konabot/plugins/roll_dice/roll_dice.py

@@ -332,9 +332,10 @@ async def generate_dice_image(number: str) -> BytesIO:
     up_direction = (51 - 16, 5 - 30)  # 右上角点 - 左上角点
 
     move_distance = (up_direction[0] * (stretch_k - 1), up_direction[1] * (stretch_k - 1))
-    
+
     # 加载背景图像，保留透明通道
-    background = cv2.imread(ASSETS_PATH / "img" / "dice" / "template.png", cv2.IMREAD_UNCHANGED)
+    background = cv2.imread(str(ASSETS_PATH / "img" / "dice" / "template.png"), cv2.IMREAD_UNCHANGED)
+    assert background is not None
 
     height, width = background.shape[:2]
 
@@ -352,7 +353,7 @@ async def generate_dice_image(number: str) -> BytesIO:
     ], dtype=np.float32)
     corners[:, 0] += offset_x
     corners[:, 1] += offset_y
-
+    
 
     # 对文本图像进行3D变换（保持透明通道）
     transformed_text, transform_matrix = perspective_transform(text_image, background, corners)

konabot/plugins/simple_notify/__init__.py

@@ -45,7 +45,7 @@ class Notify(BaseModel):
 
 
 class NotifyConfigFile(BaseModel):
-    version: int = 1
+    version: int = 2
     notifies: list[Notify] = []
     unsent: list[Notify] = []
 
@@ -89,13 +89,17 @@ async def notify_now(notify: Notify):
         if notify.target_env is None:
             await bot.send_private_msg(
                 user_id=int(notify.target),
-                message=f"代办通知：{notify.notify_msg}",
+                message=cast(Any, await UniMessage.text(f"代办通知：{notify.notify_msg}").export(
+                    bot=bot,
+                )),
             )
         else:
             await bot.send_group_msg(
                 group_id=int(notify.target_env),
                 message=cast(Any,
-                    await UniMessage().at(notify.target).text(f" 代办通知：{notify.notify_msg}").export()
+                    await UniMessage().at(
+                        notify.target
+                    ).text(f" 代办通知：{notify.notify_msg}").export(bot=bot)
                 ),
             )
     else:
@@ -197,11 +201,15 @@ async def _():
 
     NOTIFIED_FLAG["task_added"] = True
 
+    await asyncio.sleep(10)
     await DATA_FILE_LOCK.acquire()
     tasks = []
     cfg = load_notify_config()
-    for notify in cfg.notifies:
-        tasks.append(create_notify_task(notify, fail2remove=False))
+    if cfg.version == 1:
+        cfg.version = 2
+    else:
+        for notify in cfg.notifies:
+            tasks.append(create_notify_task(notify, fail2remove=False))
     DATA_FILE_LOCK.release()
 
     await asyncio.gather(*tasks)

konabot/plugins/ytpgif/__init__.py

@@ -15,14 +15,14 @@ from nonebot_plugin_alconna import (
 
 __plugin_meta__ = PluginMetadata(
     name="ytpgif",
-    description="生成来回镜像翻转的动图：动图按时间分段播放，静态图高频翻转（带参数范围保护）",
-    usage="/ytpgif [倍速=1.0] （倍速范围：0.1～20.0）",
+    description="生成来回镜像翻转的仿 YTPMV 动图。",
+    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
@@ -54,12 +54,15 @@ ytpgif_cmd = on_alconna(
 
 
 async def get_image_url(event: BaseEvent) -> Optional[str]:
+    """从事件中提取图片 URL，支持直接消息和回复"""
     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:
+        reply_msg = reply.message
+        for seg in reply_msg:
             if seg.type == "image" and seg.data.get("url"):
                 return str(seg.data["url"])
     return None
@@ -74,9 +77,11 @@ async def download_image(url: str) -> bytes:
 
 
 def resize_frame(frame: Image.Image) -> Image.Image:
+    """缩放图像，保持宽高比，不超过 MAX_SIZE"""
     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)
@@ -96,14 +101,15 @@ async def handle_ytpgif(event: BaseEvent, speed: float = 1.0):
     if not img_url:
         await ytpgif_cmd.send(
             await UniMessage.text(
-                "请发送一张图片并使用 /ytpgif，或回复一张图片来生成镜像动图。"
+                "请发送一张图片或回复一张图片来生成镜像动图。"
             ).export()
         )
         return
 
     try:
         image_data = await download_image(img_url)
-    except Exception:
+    except Exception as e:
+        print(f"[YTPGIF] 下载失败: {e}")
         await ytpgif_cmd.send(
             await UniMessage.text("❌ 图片下载失败，请重试。").export()
         )
@@ -119,22 +125,47 @@ async def handle_ytpgif(event: BaseEvent, speed: float = 1.0):
             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
+            # === 判断是否为动图 ===
+            try:
+                n_frames = getattr(src_img, "n_frames", 1)
+                is_animated = n_frames > 1
+            except Exception:
+                is_animated = False
 
             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)
+                palette = src_img.getpalette()
+
+                for idx in range(n_frames):
+                    src_img.seek(idx)
+                    frame = src_img.copy()
+                    # 检查是否需要透明通道
+                    has_alpha = (
+                        frame.mode in ("RGBA", "LA")
+                        or (frame.mode == "P" and "transparency" in frame.info)
+                    )
+                    if has_alpha:
+                        frame = frame.convert("RGBA")
+                    else:
+                        frame = frame.convert("RGB")
+                    resized_frame = resize_frame(frame)
+
+                    # 若原图有调色板，尝试保留（可选）
+                    if palette and resized_frame.mode == "P":
+                        try:
+                            resized_frame.putpalette(palette)
+                        except Exception:  # noqa
+                            pass
+
                     ms = frame.info.get("duration", int(BASE_SEGMENT_DURATION * 1000))
-                    dur_sec = max(0.01, ms / 1000.0)  # 至少 10ms
+                    dur_sec = max(0.01, ms / 1000.0)
                     frames_with_duration.append((resized_frame, dur_sec))
 
-                max_dur = BASE_SEGMENT_DURATION * speed  # 每段最大播放时间
+                max_dur = BASE_SEGMENT_DURATION * speed
                 accumulated = 0.0
                 frame_count = 0
 
@@ -167,10 +198,10 @@ async def handle_ytpgif(event: BaseEvent, speed: float = 1.0):
 
             else:
                 # === 静态图模式：制作翻转动画 ===
-                raw_frame = src_img.convert("RGB")
+                raw_frame = src_img.convert("RGBA")
                 resized_frame = resize_frame(raw_frame)
 
-                interval_sec = max(0.025, min(2.5, BASE_INTERVAL / speed))  # 限制在 25ms ~ 2.5s
+                interval_sec = max(0.025, min(2.5, BASE_INTERVAL / speed))
                 duration_ms = int(interval_sec * 1000)
 
                 frame1 = resized_frame
@@ -185,16 +216,38 @@ async def handle_ytpgif(event: BaseEvent, speed: float = 1.0):
                 )
                 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,
-            )
+            # === 🔐 关键修复：防止无透明图的颜色被当成透明 ===
+            need_transparency = False
+            for frame in output_frames:
+                if frame.mode == "RGBA":
+                    alpha_channel = frame.getchannel("A")
+                    if any(pix < 255 for pix in alpha_channel.getdata()):
+                        need_transparency = True
+                        break
+                elif frame.mode == "P" and "transparency" in frame.info:
+                    need_transparency = True
+                    break
+
+            # 如果不需要透明，则统一转为 RGB 避免调色板污染
+            if not need_transparency:
+                output_frames = [f.convert("RGB") for f in output_frames]
+
+            # 构建保存参数
+            save_kwargs = {
+                "save_all": True,
+                "append_images": output_frames[1:],
+                "format": "GIF",
+                "loop": 0,           # 无限循环
+                "duration": output_durations_ms,
+                "disposal": 2,       # 清除到背景色，避免残留
+                "optimize": False,   # 关闭抖动（等效 -dither none）
+            }
+
+            # 只有真正需要透明时才启用 transparency
+            if need_transparency:
+                save_kwargs["transparency"] = 0
+
+            output_frames[0].save(output_path, **save_kwargs)
 
         # 发送结果
         with open(output_path, "rb") as f:
@@ -204,9 +257,12 @@ async def handle_ytpgif(event: BaseEvent, speed: float = 1.0):
     except Exception as e:
         print(f"[YTPGIF] 处理失败: {e}")
         await ytpgif_cmd.send(
-            await UniMessage.text("❌ 处理失败，可能是图片格式不支持或文件过大。").export()
+            await UniMessage.text("❌ 处理失败，可能是图片格式不支持、文件损坏或过大。").export()
         )
     finally:
         for path in filter(None, [input_path, output_path]):
             if os.path.exists(path):
-                os.unlink(path)
+                try:
+                    os.unlink(path)
+                except:  # noqa
+                    pass

poetry.lock

@@ -2460,6 +2460,25 @@ urllib3 = ">=1.21.1,<3"
 socks = ["PySocks (>=1.5.6,!=1.5.7)"]
 use-chardet-on-py3 = ["chardet (>=3.0.2,<6)"]
 
+[[package]]
+name = "returns"
+version = "0.26.0"
+description = "Make your functions return something meaningful, typed, and safe!"
+optional = false
+python-versions = "<4.0,>=3.10"
+groups = ["main"]
+files = [
+    {file = "returns-0.26.0-py3-none-any.whl", hash = "sha256:7cae94c730d6c56ffd9d0f583f7a2c0b32cfe17d141837150c8e6cff3eb30d71"},
+    {file = "returns-0.26.0.tar.gz", hash = "sha256:180320e0f6e9ea9845330ccfc020f542330f05b7250941d9b9b7c00203fcc3da"},
+]
+
+[package.dependencies]
+typing-extensions = ">=4.0,<5.0"
+
+[package.extras]
+check-laws = ["hypothesis (>=6.136,<7.0)", "pytest (>=8.0,<9.0)"]
+compatible-mypy = ["mypy (>=1.12,<1.18)"]
+
 [[package]]
 name = "rich"
 version = "14.1.0"
@@ -3162,4 +3181,4 @@ type = ["pytest-mypy"]
 [metadata]
 lock-version = "2.1"
 python-versions = ">=3.12,<4.0"
-content-hash = "673703a789248d0f7369999c364352eb12f8bb5830a8b4b6918f8bab6425a763"
+content-hash = "927913b9030d1f6c126bb2d12eab7307dc6297f259c7c62e3033706457d27ce0"

pyproject.toml

@@ -21,6 +21,7 @@ dependencies = [
     "pillow (>=11.3.0,<12.0.0)",
     "imagetext-py (>=2.2.0,<3.0.0)",
     "opencv-python-headless (>=4.12.0.88,<5.0.0.0)",
+    "returns (>=0.26.0,<0.27.0)",
 ]