konabot/konabot/plugins/image_process/__init__.py

import re
from io import BytesIO

import PIL.Image
from nonebot import on_message
from nonebot.adapters import Bot
from nonebot_plugin_alconna import Alconna, Args, Image, Option, UniMessage, on_alconna

from konabot.common.nb.exc import BotExceptionMessage
from konabot.common.nb.extract_image import PIL_Image
from konabot.common.nb.match_keyword import match_keyword
from konabot.common.nb.reply_image import reply_image

cmd_black_white = on_message(rule=match_keyword("黑白"))


@cmd_black_white.handle()
async def _(img: PIL_Image, bot: Bot):
    await reply_image(cmd_black_white, bot, img.convert("LA"))


def parse_timestamp(tx: str) -> float | None:
    res = 0.0
    for component in tx.split(":"):
        res *= 60
        if not re.match(r"^\d+(\.\d+)?$", component):
            return
        res += float(component)
    return res


cmd_giftool = on_alconna(
    Alconna(
        "giftool",
        Args["img", Image | None],
        Option("--ss", Args["start_point", str]),
        Option("--frames:v", Args["frame_count", int]),
        Option("-t", Args["length", str]),
        Option("-to", Args["end_point", str]),
        Option("--speed", Args["speed_factor", float], default=1.0, alias=["-s"]),
    )
)


@cmd_giftool.handle()
async def _(
    image: PIL_Image,
    start_point: str | None = None,
    frame_count: int | None = None,
    length: str | None = None,
    speed_factor: float = 1.0,
    end_point: str | None = None,
):
    ss: None | float = None
    if start_point:
        ss = parse_timestamp(start_point)
        if ss is None:
            raise BotExceptionMessage("--ss 的格式不满足条件")

    t: None | float = None
    if length:
        t = parse_timestamp(length)
        if t is None:
            raise BotExceptionMessage("-t 的格式不满足条件")

    to: None | float = None
    if end_point:
        to = parse_timestamp(end_point)
        if to is None:
            raise BotExceptionMessage("-to 的格式不满足条件")

    if to is not None and ss is not None and to <= ss:
        raise BotExceptionMessage("错误：出点时间小于入点")
    if frame_count is not None and frame_count <= 0:
        raise BotExceptionMessage("错误：帧数量应该大于 0")
    if speed_factor == 0:
        raise BotExceptionMessage("错误：速度不能为 0")

    is_rev = speed_factor < 0
    speed_factor = abs(speed_factor)

    if not getattr(image, "is_animated", False):
        raise BotExceptionMessage("错误：输入的不是动图（GIF）")

    ##
    # 从这里开始，采样整个 GIF 图
    frames: list[PIL.Image.Image] = []
    durations: list[float] = []
    try:
        for i in range(getattr(image, "n_frames")):
            image.seek(i)
            frames.append(image.copy())
            duration = image.info.get("duration", 100) / 1000
            durations.append(duration)
    except EOFError:
        pass
    if not frames:
        raise BotExceptionMessage("错误：读取 GIF 帧失败")
    # 采样结束

    ##
    # 根据开始、结束时间或者帧数量来裁取 GIF 图

    begin_time = ss or 0
    end_time = sum(durations)
    end_time = min(begin_time + (t or end_time), to or end_time, end_time)

    accumulated = 0.0
    status = 0

    sel_frames: list[PIL.Image.Image] = []
    sel_durations: list[float] = []

    for i in range(len(frames)):
        frame = frames[i]
        duration = durations[i]

        if status == 0:
            if accumulated + duration > begin_time:
                status = 1
                sel_frames.append(frame)
                sel_durations.append(accumulated + duration - begin_time)
        elif status == 1:
            if accumulated + duration > end_time:
                sel_frames.append(frame)
                sel_durations.append(end_time - accumulated)
                break
            sel_frames.append(frame)
            sel_durations.append(duration)

        accumulated += duration

    ##
    # 加速！
    sel_durations = [dur / speed_factor * 1000 for dur in durations]

    rframes = []
    rdur = []

    acc_mod_20 = 0

    for i in range(len(sel_frames)):
        fr = sel_frames[i]
        du = round(sel_durations[i])

        if du >= 20:
            rframes.append(fr)
            rdur.append(int(du))
            acc_mod_20 = 0
        else:
            if acc_mod_20 == 0:
                rframes.append(fr)
                rdur.append(20)
                acc_mod_20 += du
            else:
                acc_mod_20 += du
                if acc_mod_20 >= 20:
                    acc_mod_20 = 0

    if len(rframes) == 1 and len(sel_frames) > 1:
        rframes.append(sel_frames[max(2, len(sel_frames) // 2)])
        rdur.append(20)

    ##
    # 收尾：看看透明度这块
    transparency_flag = False
    for f in rframes:
        if f.mode == "RGBA":
            if any(pix < 255 for pix in f.getchannel("A").getdata()):
                transparency_flag = True
                break
        elif f.mode == "P" and "transparency" in f.info:
            transparency_flag = True
            break

    tf = {}
    if transparency_flag:
        tf["transparency"] = 0

    if is_rev:
        rframes = rframes[::-1]
        rdur = rdur[::-1]

    output_img = BytesIO()
    if rframes:
        rframes[0].save(
            output_img,
            format="GIF",
            save_all=True,
            append_images=rframes[1:],
            duration=rdur,
            loop=0,
            optimize=False,
            disposal=2,
            **tf,
        )
    else:
        raise BotExceptionMessage("错误：没有可输出的帧")
    output_img.seek(0)

    await cmd_giftool.send(await UniMessage().image(raw=output_img).export())