63 lines
1.8 KiB
Python
63 lines
1.8 KiB
Python
 
|
"""GIF 倒放 —— 业务逻辑(Service 层)。"""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from io import BytesIO
|
||
|
|
|
||
|
|
import httpx
|
||
|
|
from PIL import Image, ImageSequence
|
||
|
|
from nonebot.log import logger
|
||
|
|
|
||
|
|
|
||
|
|
async def reverse_gif_bytes(image_urls: list[str]) -> tuple[bytes | None, str | None]:
|
||
|
|
"""下载并翻转 GIF。
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
(bytes, None) 成功；
|
||
|
|
(None, msg) 失败原因(无 GIF / 帧数过多 / 处理失败)。
|
||
|
|
"""
|
||
|
|
gif_img = []
|
||
|
|
async with httpx.AsyncClient(timeout=30) as client:
|
||
|
|
for image_url in image_urls:
|
||
|
|
try:
|
||
|
|
resp = await client.get(image_url)
|
||
|
|
resp.raise_for_status()
|
||
|
|
image = Image.open(BytesIO(resp.content))
|
||
|
|
if getattr(image, "n_frames", 1) == 1:
|
||
|
|
continue
|
||
|
|
gif_img.append(image)
|
||
|
|
except Exception as e: # noqa: BLE001
|
||
|
|
logger.warning(f"下载或解析图片失败: {e}")
|
||
|
|
|
||
|
|
if not gif_img:
|
||
|
|
return None, "没有发现GIF图片"
|
||
|
|
|
||
|
|
suit_gifs = [gif for gif in gif_img if gif.n_frames <= 400]
|
||
|
|
if not suit_gifs:
|
||
|
|
return None, "GIF帧数太多了，懒得倒放😅"
|
||
|
|
|
||
|
|
suit_image = suit_gifs[0]
|
||
|
|
sequence = [f.copy() for f in ImageSequence.Iterator(suit_image)]
|
||
|
|
logger.info(f"GIF帧数: {len(sequence)}")
|
||
|
|
if len(sequence) <= 2:
|
||
|
|
return None, "写入文件时发生未知错误"
|
||
|
|
|
||
|
|
sequence.reverse()
|
||
|
|
bio = BytesIO()
|
||
|
|
sequence[0].save(
|
||
|
|
bio,
|
||
|
|
format="GIF",
|
||
|
|
save_all=True,
|
||
|
|
append_images=sequence[1:],
|
||
|
|
disposal=1,
|
||
|
|
loop=0,
|
||
|
|
)
|
||
|
|
bio.seek(0)
|
||
|
|
return bio.getvalue(), None
|
||
|
|
|
||
|
|
|
||
|
|
def image_to_bytes(image: Image.Image, input_format: str) -> bytes:
|
||
|
|
with BytesIO() as bio:
|
||
|
|
image.save(bio, format=input_format)
|
||
|
|
return bio.getvalue()
|