From 715f62af584970c9dbc572da262535c376d7e1c5 Mon Sep 17 00:00:00 2001 From: xz-dev Date: Tue, 16 Jul 2024 16:07:54 +0800 Subject: [PATCH 01/10] feat: support animated(webm) sticker --- requirements.txt | 1 + sticker/lib/util.py | 62 +++++++++++++++++++++++++++++++++++++--- sticker/pack.py | 6 ++-- sticker/stickerimport.py | 6 ++-- 4 files changed, 65 insertions(+), 10 deletions(-) diff --git a/requirements.txt b/requirements.txt index 6d89dbf..bd4cfb0 100644 --- a/requirements.txt +++ b/requirements.txt @@ -4,3 +4,4 @@ pillow telethon cryptg python-magic +moviepy diff --git a/sticker/lib/util.py b/sticker/lib/util.py index 2b3fe2a..43e01d3 100644 --- a/sticker/lib/util.py +++ b/sticker/lib/util.py @@ -17,14 +17,50 @@ from functools import partial from io import BytesIO import os.path import json +import tempfile +import mimetypes +try: + import magic +except ImportError: + print("[Warning] Magic is not installed, using file extensions to guess mime types") + magic = None from PIL import Image from . import matrix open_utf8 = partial(open, encoding='UTF-8') -def convert_image(data: bytes) -> (bytes, int, int): + +def guess_mime(data: bytes) -> str: + mime = None + if magic: + try: + return magic.Magic(mime=True).from_buffer(data) + except Exception: + pass + else: + with tempfile.NamedTemporaryFile(delete=False) as temp: + temp.write(data) + temp.close() + mime, _ = mimetypes.guess_type(temp.name) + return mime or "image/png" + + +def video_to_gif(data: bytes, mime: str) -> bytes: + from moviepy.editor import VideoFileClip + ext = mimetypes.guess_extension(mime) + with tempfile.NamedTemporaryFile(suffix=ext) as temp: + temp.write(data) + temp.flush() + with tempfile.NamedTemporaryFile(suffix=".gif") as gif: + clip = VideoFileClip(temp.name) + clip.write_gif(gif.name, logger=None) + gif.seek(0) + return gif.read() + + +def _convert_image(data: bytes) -> (bytes, int, int): image: Image.Image = Image.open(BytesIO(data)).convert("RGBA") new_file = BytesIO() image.save(new_file, "png") @@ -40,6 +76,24 @@ def convert_image(data: bytes) -> (bytes, int, int): return new_file.getvalue(), w, h +def convert_image(data: bytes) -> (bytes, str, int, int): + mimetype = guess_mime(data) + if mimetype.startswith("video/"): + data = video_to_gif(data, mimetype) + print(".", end="", flush=True) + mimetype = "image/gif" + try: + rlt = _convert_image(data) + return rlt[0], mimetype, rlt[1], rlt[2] + except Exception as e: + print(f"Error converting image, mimetype: {mimetype}") + ext = mimetypes.guess_extension(mimetype) + with tempfile.NamedTemporaryFile(delete=False, suffix=ext) as temp: + temp.write(data) + print(f"Saved to {temp.name}") + raise e + + def add_to_index(name: str, output_dir: str) -> None: index_path = os.path.join(output_dir, "index.json") try: @@ -57,7 +111,7 @@ def add_to_index(name: str, output_dir: str) -> None: def make_sticker(mxc: str, width: int, height: int, size: int, - body: str = "") -> matrix.StickerInfo: + mimetype: str, body: str = "") -> matrix.StickerInfo: return { "body": body, "url": mxc, @@ -65,7 +119,7 @@ def make_sticker(mxc: str, width: int, height: int, size: int, "w": width, "h": height, "size": size, - "mimetype": "image/png", + "mimetype": mimetype, # Element iOS compatibility hack "thumbnail_url": mxc, @@ -73,7 +127,7 @@ def make_sticker(mxc: str, width: int, height: int, size: int, "w": width, "h": height, "size": size, - "mimetype": "image/png", + "mimetype": mimetype, }, }, "msgtype": "m.sticker", diff --git a/sticker/pack.py b/sticker/pack.py index f082370..6b1a646 100644 --- a/sticker/pack.py +++ b/sticker/pack.py @@ -77,11 +77,11 @@ async def upload_sticker(file: str, directory: str, old_stickers: Dict[str, matr } print(f".. using existing upload") else: - image_data, width, height = util.convert_image(image_data) + image_data, mimetype, width, height = util.convert_image(image_data) print(".", end="", flush=True) - mxc = await matrix.upload(image_data, "image/png", file) + mxc = await matrix.upload(image_data, mimetype, file) print(".", end="", flush=True) - sticker = util.make_sticker(mxc, width, height, len(image_data), name) + sticker = util.make_sticker(mxc, width, height, len(image_data), mimetype, name) sticker["id"] = sticker_id print(" uploaded", flush=True) return sticker diff --git a/sticker/stickerimport.py b/sticker/stickerimport.py index 534f3c4..6b12961 100644 --- a/sticker/stickerimport.py +++ b/sticker/stickerimport.py @@ -33,11 +33,11 @@ async def reupload_document(client: TelegramClient, document: Document) -> matri print(f"Reuploading {document.id}", end="", flush=True) data = await client.download_media(document, file=bytes) print(".", end="", flush=True) - data, width, height = util.convert_image(data) + data, mimetype, width, height = util.convert_image(data) print(".", end="", flush=True) - mxc = await matrix.upload(data, "image/png", f"{document.id}.png") + mxc = await matrix.upload(data, mimetype, f"{document.id}.png") print(".", flush=True) - return util.make_sticker(mxc, width, height, len(data)) + return util.make_sticker(mxc, width, height, len(data), mimetype) def add_meta(document: Document, info: matrix.StickerInfo, pack: StickerSetFull) -> None: From be477874e3b66cc5f5432b1fb1c684db543c70c8 Mon Sep 17 00:00:00 2001 From: xz-dev Date: Tue, 16 Jul 2024 18:15:49 +0800 Subject: [PATCH 02/10] feat: support animated(lottie) sticker --- requirements.txt | 1 + sticker/lib/util.py | 32 +++++++++++++++++++++++++++++--- sticker/pack.py | 2 +- sticker/stickerimport.py | 2 +- 4 files changed, 32 insertions(+), 5 deletions(-) diff --git a/requirements.txt b/requirements.txt index bd4cfb0..2b3edb0 100644 --- a/requirements.txt +++ b/requirements.txt @@ -5,3 +5,4 @@ telethon cryptg python-magic moviepy +lottie[all] diff --git a/sticker/lib/util.py b/sticker/lib/util.py index 43e01d3..ec96a2b 100644 --- a/sticker/lib/util.py +++ b/sticker/lib/util.py @@ -76,16 +76,42 @@ def _convert_image(data: bytes) -> (bytes, int, int): return new_file.getvalue(), w, h -def convert_image(data: bytes) -> (bytes, str, int, int): +def _convert_sticker(data: bytes) -> (bytes, str, int, int): mimetype = guess_mime(data) if mimetype.startswith("video/"): data = video_to_gif(data, mimetype) print(".", end="", flush=True) mimetype = "image/gif" + elif mimetype.startswith("application/gzip"): + print(".", end="", flush=True) + # unzip file + import gzip + with gzip.open(BytesIO(data), "rb") as f: + data = f.read() + mimetype = guess_mime(data) + suffix = mimetypes.guess_extension(mimetype) + with tempfile.NamedTemporaryFile(suffix=suffix) as temp: + temp.write(data) + with tempfile.NamedTemporaryFile(suffix=".gif") as gif: + # run lottie_convert.py input output + print(".", end="", flush=True) + import subprocess + cmd = ["lottie_convert.py", temp.name, gif.name] + result = subprocess.run(cmd, capture_output=True, text=True) + if result.returncode != 0: + raise RuntimeError(f"Run {cmd} failed with code {retcode}, Error occurred:\n{result.stderr}") + gif.seek(0) + data = gif.read() + mimetype = "image/gif" + rlt = _convert_image(data) + return rlt[0], mimetype, rlt[1], rlt[2] + + +def convert_sticker(data: bytes) -> (bytes, str, int, int): try: - rlt = _convert_image(data) - return rlt[0], mimetype, rlt[1], rlt[2] + return _convert_sticker(data) except Exception as e: + mimetype = guess_mime(data) print(f"Error converting image, mimetype: {mimetype}") ext = mimetypes.guess_extension(mimetype) with tempfile.NamedTemporaryFile(delete=False, suffix=ext) as temp: diff --git a/sticker/pack.py b/sticker/pack.py index 6b1a646..4815bc5 100644 --- a/sticker/pack.py +++ b/sticker/pack.py @@ -77,7 +77,7 @@ async def upload_sticker(file: str, directory: str, old_stickers: Dict[str, matr } print(f".. using existing upload") else: - image_data, mimetype, width, height = util.convert_image(image_data) + image_data, mimetype, width, height = util.convert_sticker(image_data) print(".", end="", flush=True) mxc = await matrix.upload(image_data, mimetype, file) print(".", end="", flush=True) diff --git a/sticker/stickerimport.py b/sticker/stickerimport.py index 6b12961..6d1e7be 100644 --- a/sticker/stickerimport.py +++ b/sticker/stickerimport.py @@ -33,7 +33,7 @@ async def reupload_document(client: TelegramClient, document: Document) -> matri print(f"Reuploading {document.id}", end="", flush=True) data = await client.download_media(document, file=bytes) print(".", end="", flush=True) - data, mimetype, width, height = util.convert_image(data) + data, mimetype, width, height = util.convert_sticker(data) print(".", end="", flush=True) mxc = await matrix.upload(data, mimetype, f"{document.id}.png") print(".", flush=True) From a6b8c093797b08cf6954e0a97b553d9aaf50fbf6 Mon Sep 17 00:00:00 2001 From: xz-dev Date: Tue, 16 Jul 2024 18:58:10 +0800 Subject: [PATCH 03/10] fix: support animated sticker convert(RGBA) --- sticker/lib/util.py | 35 +++++++++++++++++++++++++++++------ 1 file changed, 29 insertions(+), 6 deletions(-) diff --git a/sticker/lib/util.py b/sticker/lib/util.py index ec96a2b..b2f5bab 100644 --- a/sticker/lib/util.py +++ b/sticker/lib/util.py @@ -25,7 +25,7 @@ try: except ImportError: print("[Warning] Magic is not installed, using file extensions to guess mime types") magic = None -from PIL import Image +from PIL import Image, ImageSequence from . import matrix @@ -60,11 +60,33 @@ def video_to_gif(data: bytes, mime: str) -> bytes: return gif.read() -def _convert_image(data: bytes) -> (bytes, int, int): - image: Image.Image = Image.open(BytesIO(data)).convert("RGBA") +def _convert_image(data: bytes, mimetype: str) -> (bytes, int, int): + image: Image.Image = Image.open(BytesIO(data)) new_file = BytesIO() - image.save(new_file, "png") - w, h = image.size + suffix = mimetypes.guess_extension(mimetype) + if suffix: + suffix = suffix[1:] + # Determine if the image is a GIF + is_animated = getattr(image, "is_animated", False) + if is_animated: + frames = [frame.convert("RGBA") for frame in ImageSequence.Iterator(image)] + # Save the new GIF + frames[0].save( + new_file, + format='GIF', + save_all=True, + append_images=frames[1:], + loop=image.info.get('loop', 0), # Default loop to 0 if not present + duration=image.info.get('duration', 100), # Set a default duration if not present + transparency=image.info.get('transparency', 255), # Default to 255 if transparency is not present + disposal=image.info.get('disposal', 2) # Default to disposal method 2 (restore to background) + ) + # Get the size of the first frame to determine resizing + w, h = frames[0].size + else: + image = image.convert("RGBA") + image.save(new_file, format=suffix) + w, h = image.size if w > 256 or h > 256: # Set the width and height to lower values so clients wouldn't show them as huge images if w > h: @@ -103,7 +125,8 @@ def _convert_sticker(data: bytes) -> (bytes, str, int, int): gif.seek(0) data = gif.read() mimetype = "image/gif" - rlt = _convert_image(data) + rlt = _convert_image(data, mimetype) + suffix = mimetypes.guess_extension(mimetype) return rlt[0], mimetype, rlt[1], rlt[2] From e38090e95231ba43242c873959bba6c40840f56e Mon Sep 17 00:00:00 2001 From: xz-dev Date: Tue, 16 Jul 2024 21:17:07 +0800 Subject: [PATCH 04/10] fix: fix webm duration via ffmpeg --- sticker/lib/util.py | 18 ++++++++++++++++-- 1 file changed, 16 insertions(+), 2 deletions(-) diff --git a/sticker/lib/util.py b/sticker/lib/util.py index b2f5bab..b2796bd 100644 --- a/sticker/lib/util.py +++ b/sticker/lib/util.py @@ -40,7 +40,7 @@ def guess_mime(data: bytes) -> str: except Exception: pass else: - with tempfile.NamedTemporaryFile(delete=False) as temp: + with tempfile.NamedTemporaryFile() as temp: temp.write(data) temp.close() mime, _ = mimetypes.guess_type(temp.name) @@ -48,12 +48,26 @@ def guess_mime(data: bytes) -> str: def video_to_gif(data: bytes, mime: str) -> bytes: - from moviepy.editor import VideoFileClip ext = mimetypes.guess_extension(mime) + if mime.startswith("video/"): + # run ffmpeg to fix duration + with tempfile.NamedTemporaryFile(suffix=ext) as temp: + temp.write(data) + temp.flush() + with tempfile.NamedTemporaryFile(suffix=ext) as temp_fixed: + import subprocess + print(".", end="", flush=True) + result = subprocess.run(["ffmpeg", "-y", "-i", temp.name, "-codec", "copy", temp_fixed.name], + capture_output=True) + if result.returncode != 0: + raise RuntimeError(f"Run ffmpeg failed with code {result.returncode}, Error occurred:\n{result.stderr}") + temp_fixed.seek(0) + data = temp_fixed.read() with tempfile.NamedTemporaryFile(suffix=ext) as temp: temp.write(data) temp.flush() with tempfile.NamedTemporaryFile(suffix=".gif") as gif: + from moviepy.editor import VideoFileClip clip = VideoFileClip(temp.name) clip.write_gif(gif.name, logger=None) gif.seek(0) From 47a98ba81b68076c578d88fb627acd4f0c0ca1ca Mon Sep 17 00:00:00 2001 From: xz-dev Date: Tue, 16 Jul 2024 21:56:00 +0800 Subject: [PATCH 05/10] perf: optimize gif via gifsicle --- sticker/lib/util.py | 22 +++++++++++++++++++--- 1 file changed, 19 insertions(+), 3 deletions(-) diff --git a/sticker/lib/util.py b/sticker/lib/util.py index b2796bd..e84b450 100644 --- a/sticker/lib/util.py +++ b/sticker/lib/util.py @@ -16,6 +16,7 @@ from functools import partial from io import BytesIO import os.path +import subprocess import json import tempfile import mimetypes @@ -55,7 +56,6 @@ def video_to_gif(data: bytes, mime: str) -> bytes: temp.write(data) temp.flush() with tempfile.NamedTemporaryFile(suffix=ext) as temp_fixed: - import subprocess print(".", end="", flush=True) result = subprocess.run(["ffmpeg", "-y", "-i", temp.name, "-codec", "copy", temp_fixed.name], capture_output=True) @@ -74,6 +74,19 @@ def video_to_gif(data: bytes, mime: str) -> bytes: return gif.read() +def opermize_gif(data: bytes) -> bytes: + with tempfile.NamedTemporaryFile() as gif: + gif.write(data) + gif.flush() + # use gifsicle to optimize gif + result = subprocess.run(["gifsicle", "--batch", "--optimize=3", "--colors=256", gif.name], + capture_output=True) + if result.returncode != 0: + raise RuntimeError(f"Run gifsicle failed with code {result.returncode}, Error occurred:\n{result.stderr}") + gif.seek(0) + return gif.read() + + def _convert_image(data: bytes, mimetype: str) -> (bytes, int, int): image: Image.Image = Image.open(BytesIO(data)) new_file = BytesIO() @@ -140,8 +153,11 @@ def _convert_sticker(data: bytes) -> (bytes, str, int, int): data = gif.read() mimetype = "image/gif" rlt = _convert_image(data, mimetype) - suffix = mimetypes.guess_extension(mimetype) - return rlt[0], mimetype, rlt[1], rlt[2] + data = rlt[0] + if mimetype == "image/gif": + print(".", end="", flush=True) + data = opermize_gif(data) + return data, mimetype, rlt[1], rlt[2] def convert_sticker(data: bytes) -> (bytes, str, int, int): From 8ce5be04fb5e0e2e7b665aeca3e717424ac1149d Mon Sep 17 00:00:00 2001 From: xz-dev Date: Sat, 14 Sep 2024 16:44:32 +0800 Subject: [PATCH 06/10] perf: ignore imported resources --- .gitignore | 2 ++ 1 file changed, 2 insertions(+) diff --git a/.gitignore b/.gitignore index b2a42f7..3e0fad4 100644 --- a/.gitignore +++ b/.gitignore @@ -5,9 +5,11 @@ *.pyc __pycache__ *.egg-info +build/ node_modules web/lib/import-map.json +web/packs/*.json *.session /*.json From a83bc15208f6ddf59b555c8f30c0e6496cda2a89 Mon Sep 17 00:00:00 2001 From: xz-dev Date: Sat, 14 Sep 2024 23:03:26 +0800 Subject: [PATCH 07/10] fix: keep webm transparency --- requirements.txt | 1 - sticker/lib/util.py | 42 +++++++++++++++++++++++++++++++++++------- 2 files changed, 35 insertions(+), 8 deletions(-) diff --git a/requirements.txt b/requirements.txt index 2b3edb0..560d2a5 100644 --- a/requirements.txt +++ b/requirements.txt @@ -4,5 +4,4 @@ pillow telethon cryptg python-magic -moviepy lottie[all] diff --git a/sticker/lib/util.py b/sticker/lib/util.py index e84b450..3e4432e 100644 --- a/sticker/lib/util.py +++ b/sticker/lib/util.py @@ -48,6 +48,32 @@ def guess_mime(data: bytes) -> str: return mime or "image/png" +def video_to_webp(data: bytes) -> bytes: + mime = guess_mime(data) + ext = mimetypes.guess_extension(mime) + with tempfile.NamedTemporaryFile(suffix=ext) as video: + video.write(data) + video.flush() + with tempfile.NamedTemporaryFile(suffix=".webp") as webp: + print(".", end="", flush=True) + ffmpeg_encoder_args = [] + if mime == "video/webm": + encode = subprocess.run(["ffprobe", "-v", "error", "-select_streams", "v:0", "-show_entries", "stream=codec_name", "-of", "default=nokey=1:noprint_wrappers=1", video.name], capture_output=True, text=True).stdout.strip() + ffmpeg_encoder = None + if encode == "vp8": + ffmpeg_encoder = "libvpx" + elif encode == "vp9": + ffmpeg_encoder = "libvpx-vp9" + if ffmpeg_encoder: + ffmpeg_encoder_args = ["-c:v", ffmpeg_encoder] + result = subprocess.run(["ffmpeg", "-y", "-threads", "auto", *ffmpeg_encoder_args, "-i", video.name, "-lossless", "1", webp.name], + capture_output=True) + if result.returncode != 0: + raise RuntimeError(f"Run ffmpeg failed with code {result.returncode}, Error occurred:\n{result.stderr}") + webp.seek(0) + return webp.read() + + def video_to_gif(data: bytes, mime: str) -> bytes: ext = mimetypes.guess_extension(mime) if mime.startswith("video/"): @@ -57,19 +83,21 @@ def video_to_gif(data: bytes, mime: str) -> bytes: temp.flush() with tempfile.NamedTemporaryFile(suffix=ext) as temp_fixed: print(".", end="", flush=True) - result = subprocess.run(["ffmpeg", "-y", "-i", temp.name, "-codec", "copy", temp_fixed.name], + result = subprocess.run(["ffmpeg", "-y", "-threads", "auto", "-i", temp.name, "-codec", "copy", temp_fixed.name], capture_output=True) if result.returncode != 0: raise RuntimeError(f"Run ffmpeg failed with code {result.returncode}, Error occurred:\n{result.stderr}") temp_fixed.seek(0) data = temp_fixed.read() + data = video_to_webp(data) with tempfile.NamedTemporaryFile(suffix=ext) as temp: temp.write(data) temp.flush() with tempfile.NamedTemporaryFile(suffix=".gif") as gif: - from moviepy.editor import VideoFileClip - clip = VideoFileClip(temp.name) - clip.write_gif(gif.name, logger=None) + print(".", end="", flush=True) + im = Image.open(temp.name) + im.info.pop('background', None) + im.save(gif.name, save_all=True, lossless=True, quality=100, method=6) gif.seek(0) return gif.read() @@ -105,7 +133,6 @@ def _convert_image(data: bytes, mimetype: str) -> (bytes, int, int): append_images=frames[1:], loop=image.info.get('loop', 0), # Default loop to 0 if not present duration=image.info.get('duration', 100), # Set a default duration if not present - transparency=image.info.get('transparency', 255), # Default to 255 if transparency is not present disposal=image.info.get('disposal', 2) # Default to disposal method 2 (restore to background) ) # Get the size of the first frame to determine resizing @@ -147,7 +174,8 @@ def _convert_sticker(data: bytes) -> (bytes, str, int, int): import subprocess cmd = ["lottie_convert.py", temp.name, gif.name] result = subprocess.run(cmd, capture_output=True, text=True) - if result.returncode != 0: + retcode = result.returncode + if retcode != 0: raise RuntimeError(f"Run {cmd} failed with code {retcode}, Error occurred:\n{result.stderr}") gif.seek(0) data = gif.read() @@ -157,7 +185,7 @@ def _convert_sticker(data: bytes) -> (bytes, str, int, int): if mimetype == "image/gif": print(".", end="", flush=True) data = opermize_gif(data) - return data, mimetype, rlt[1], rlt[2] + return data, mimetype, rlt[1], rlt[2] def convert_sticker(data: bytes) -> (bytes, str, int, int): From a0ef9f84be89f4b7ff6239215d16cfc63f318b7b Mon Sep 17 00:00:00 2001 From: xz-dev Date: Sun, 15 Sep 2024 13:21:41 +0800 Subject: [PATCH 08/10] perf: convert static animated WebP to PNG --- sticker/lib/util.py | 89 +++++++++++++++++++++++++++++++-------------- 1 file changed, 61 insertions(+), 28 deletions(-) diff --git a/sticker/lib/util.py b/sticker/lib/util.py index 3e4432e..580824c 100644 --- a/sticker/lib/util.py +++ b/sticker/lib/util.py @@ -15,6 +15,7 @@ # along with this program. If not, see . from functools import partial from io import BytesIO +import numpy as np import os.path import subprocess import json @@ -48,7 +49,7 @@ def guess_mime(data: bytes) -> str: return mime or "image/png" -def video_to_webp(data: bytes) -> bytes: +def _video_to_webp(data: bytes) -> bytes: mime = guess_mime(data) ext = mimetypes.guess_extension(mime) with tempfile.NamedTemporaryFile(suffix=ext) as video: @@ -74,32 +75,62 @@ def video_to_webp(data: bytes) -> bytes: return webp.read() -def video_to_gif(data: bytes, mime: str) -> bytes: +def video_to_webp(data: bytes) -> bytes: + mime = guess_mime(data) ext = mimetypes.guess_extension(mime) - if mime.startswith("video/"): - # run ffmpeg to fix duration - with tempfile.NamedTemporaryFile(suffix=ext) as temp: - temp.write(data) - temp.flush() - with tempfile.NamedTemporaryFile(suffix=ext) as temp_fixed: - print(".", end="", flush=True) - result = subprocess.run(["ffmpeg", "-y", "-threads", "auto", "-i", temp.name, "-codec", "copy", temp_fixed.name], - capture_output=True) - if result.returncode != 0: - raise RuntimeError(f"Run ffmpeg failed with code {result.returncode}, Error occurred:\n{result.stderr}") - temp_fixed.seek(0) - data = temp_fixed.read() - data = video_to_webp(data) + # run ffmpeg to fix duration with tempfile.NamedTemporaryFile(suffix=ext) as temp: temp.write(data) temp.flush() - with tempfile.NamedTemporaryFile(suffix=".gif") as gif: + with tempfile.NamedTemporaryFile(suffix=ext) as temp_fixed: print(".", end="", flush=True) - im = Image.open(temp.name) + result = subprocess.run(["ffmpeg", "-y", "-threads", "auto", "-i", temp.name, "-codec", "copy", temp_fixed.name], + capture_output=True) + if result.returncode != 0: + raise RuntimeError(f"Run ffmpeg failed with code {result.returncode}, Error occurred:\n{result.stderr}") + temp_fixed.seek(0) + data = temp_fixed.read() + return _video_to_webp(data) + + +def webp_to_others(data: bytes, mimetype: str) -> bytes: + with tempfile.NamedTemporaryFile(suffix=".webp") as webp: + webp.write(data) + webp.flush() + ext = mimetypes.guess_extension(mimetype) + with tempfile.NamedTemporaryFile(suffix=ext) as img: + print(".", end="", flush=True) + im = Image.open(webp.name) im.info.pop('background', None) - im.save(gif.name, save_all=True, lossless=True, quality=100, method=6) - gif.seek(0) - return gif.read() + im.save(img.name, save_all=True, lossless=True, quality=100, method=6) + img.seek(0) + return img.read() + + +def is_uniform_animated_webp(data: bytes) -> bool: + img = Image.open(BytesIO(data)) + if img.n_frames <= 1: + return False + + first_frame = np.array(img) + for frame_number in range(1, img.n_frames): + img.seek(frame_number) + current_frame = np.array(img) + if not np.array_equal(first_frame, current_frame): + return False + + return True + + +def webp_to_gif_or_png(data: bytes) -> bytes: + # check if the webp is animated + image: Image.Image = Image.open(BytesIO(data)) + is_animated = getattr(image, "is_animated", False) + if is_animated and is_uniform_animated_webp(data): + return webp_to_others(data, "image/gif") + else: + # convert to png + return webp_to_others(data, "image/png") def opermize_gif(data: bytes) -> bytes: @@ -118,9 +149,6 @@ def opermize_gif(data: bytes) -> bytes: def _convert_image(data: bytes, mimetype: str) -> (bytes, int, int): image: Image.Image = Image.open(BytesIO(data)) new_file = BytesIO() - suffix = mimetypes.guess_extension(mimetype) - if suffix: - suffix = suffix[1:] # Determine if the image is a GIF is_animated = getattr(image, "is_animated", False) if is_animated: @@ -138,6 +166,9 @@ def _convert_image(data: bytes, mimetype: str) -> (bytes, int, int): # Get the size of the first frame to determine resizing w, h = frames[0].size else: + suffix = mimetypes.guess_extension(mimetype) + if suffix: + suffix = suffix[1:] image = image.convert("RGBA") image.save(new_file, format=suffix) w, h = image.size @@ -155,9 +186,8 @@ def _convert_image(data: bytes, mimetype: str) -> (bytes, int, int): def _convert_sticker(data: bytes) -> (bytes, str, int, int): mimetype = guess_mime(data) if mimetype.startswith("video/"): - data = video_to_gif(data, mimetype) + data = video_to_webp(data) print(".", end="", flush=True) - mimetype = "image/gif" elif mimetype.startswith("application/gzip"): print(".", end="", flush=True) # unzip file @@ -168,7 +198,7 @@ def _convert_sticker(data: bytes) -> (bytes, str, int, int): suffix = mimetypes.guess_extension(mimetype) with tempfile.NamedTemporaryFile(suffix=suffix) as temp: temp.write(data) - with tempfile.NamedTemporaryFile(suffix=".gif") as gif: + with tempfile.NamedTemporaryFile(suffix=".webp") as gif: # run lottie_convert.py input output print(".", end="", flush=True) import subprocess @@ -179,7 +209,10 @@ def _convert_sticker(data: bytes) -> (bytes, str, int, int): raise RuntimeError(f"Run {cmd} failed with code {retcode}, Error occurred:\n{result.stderr}") gif.seek(0) data = gif.read() - mimetype = "image/gif" + mimetype = guess_mime(data) + if mimetype == "image/webp": + data = webp_to_gif_or_png(data) + mimetype = guess_mime(data) rlt = _convert_image(data, mimetype) data = rlt[0] if mimetype == "image/gif": From 86cb2edcfa4e04114514438a4c25e0e4170e093b Mon Sep 17 00:00:00 2001 From: xz-dev Date: Sun, 15 Sep 2024 16:26:58 +0800 Subject: [PATCH 09/10] fix: Improve edge transparency handling by modifying only the Alpha channel In our testing, the method of exclusively processing the Alpha channel yielded the best results. This approach focuses on adjusting transparency while preserving the RGB color information, which prevents color distortion and maintains image detail. Key reasons for the improvement include: - Protecting RGB from color alterations, avoiding color seepage and contamination. - Precisely removing unwanted semi-transparency in the Alpha channel, eliminating white edges. - Simplifying the process, reducing complexity, and minimizing risk of introducing new issues. By targeting transparency issues directly in the Alpha channel, we achieve cleaner edges without compromising the image's color quality and detail. --- sticker/lib/util.py | 46 ++++++++++++++++++++++++++++++++++++++++++--- 1 file changed, 43 insertions(+), 3 deletions(-) diff --git a/sticker/lib/util.py b/sticker/lib/util.py index 580824c..3e41bce 100644 --- a/sticker/lib/util.py +++ b/sticker/lib/util.py @@ -27,7 +27,7 @@ try: except ImportError: print("[Warning] Magic is not installed, using file extensions to guess mime types") magic = None -from PIL import Image, ImageSequence +from PIL import Image, ImageSequence, ImageFilter from . import matrix @@ -93,6 +93,32 @@ def video_to_webp(data: bytes) -> bytes: return _video_to_webp(data) +def process_frame(frame): + """ + Process GIF frame, repair edges, ensure no white or semi-transparent pixels, while keeping color information intact. + """ + frame = frame.convert('RGBA') + + # Decompose Alpha channel + alpha = frame.getchannel('A') + + # Process Alpha channel with threshold, remove semi-transparent pixels + # Threshold can be adjusted as needed (0-255), 128 is the middle value + threshold = 128 + alpha = alpha.point(lambda x: 255 if x >= threshold else 0) + + # Process Alpha channel with MinFilter, remove edge noise + alpha = alpha.filter(ImageFilter.MinFilter(3)) + + # Process Alpha channel with MaxFilter, repair edges + alpha = alpha.filter(ImageFilter.MaxFilter(3)) + + # Apply processed Alpha channel back to image + frame.putalpha(alpha) + + return frame + + def webp_to_others(data: bytes, mimetype: str) -> bytes: with tempfile.NamedTemporaryFile(suffix=".webp") as webp: webp.write(data) @@ -102,7 +128,21 @@ def webp_to_others(data: bytes, mimetype: str) -> bytes: print(".", end="", flush=True) im = Image.open(webp.name) im.info.pop('background', None) - im.save(img.name, save_all=True, lossless=True, quality=100, method=6) + + if mimetype == "image/gif": + frames = [] + duration = [] + + for frame in ImageSequence.Iterator(im): + frame = process_frame(frame) + frames.append(frame) + duration.append(frame.info.get('duration', 100)) + + frames[0].save(img.name, save_all=True, lossless=True, quality=100, method=6, + append_images=frames[1:], loop=0, duration=duration, disposal=2) + else: + im.save(img.name, save_all=True, lossless=True, quality=100, method=6) + img.seek(0) return img.read() @@ -126,7 +166,7 @@ def webp_to_gif_or_png(data: bytes) -> bytes: # check if the webp is animated image: Image.Image = Image.open(BytesIO(data)) is_animated = getattr(image, "is_animated", False) - if is_animated and is_uniform_animated_webp(data): + if is_animated and not is_uniform_animated_webp(data): return webp_to_others(data, "image/gif") else: # convert to png From ca296b41c68581b5777f2070e96412d3f716768c Mon Sep 17 00:00:00 2001 From: xz-dev Date: Sun, 15 Sep 2024 22:49:17 +0800 Subject: [PATCH 10/10] fix: is_uniform_animated_webp --- sticker/lib/util.py | 103 ++++++++++++++++++++++---------------------- 1 file changed, 52 insertions(+), 51 deletions(-) diff --git a/sticker/lib/util.py b/sticker/lib/util.py index 3e41bce..0bf6c1b 100644 --- a/sticker/lib/util.py +++ b/sticker/lib/util.py @@ -148,29 +148,30 @@ def webp_to_others(data: bytes, mimetype: str) -> bytes: def is_uniform_animated_webp(data: bytes) -> bool: - img = Image.open(BytesIO(data)) - if img.n_frames <= 1: - return False + with Image.open(BytesIO(data)) as img: + if img.n_frames <= 1: + return True - first_frame = np.array(img) - for frame_number in range(1, img.n_frames): - img.seek(frame_number) - current_frame = np.array(img) - if not np.array_equal(first_frame, current_frame): - return False + img_iter = ImageSequence.Iterator(img) + first_frame = np.array(img_iter[0].convert("RGBA")) + + for frame in img_iter: + current_frame = np.array(frame.convert("RGBA")) + if not np.array_equal(first_frame, current_frame): + return False return True def webp_to_gif_or_png(data: bytes) -> bytes: - # check if the webp is animated - image: Image.Image = Image.open(BytesIO(data)) - is_animated = getattr(image, "is_animated", False) - if is_animated and not is_uniform_animated_webp(data): - return webp_to_others(data, "image/gif") - else: - # convert to png - return webp_to_others(data, "image/png") + with Image.open(BytesIO(data)) as image: + # check if the webp is animated + is_animated = getattr(image, "is_animated", False) + if is_animated and not is_uniform_animated_webp(data): + return webp_to_others(data, "image/gif") + else: + # convert to png + return webp_to_others(data, "image/png") def opermize_gif(data: bytes) -> bytes: @@ -187,40 +188,40 @@ def opermize_gif(data: bytes) -> bytes: def _convert_image(data: bytes, mimetype: str) -> (bytes, int, int): - image: Image.Image = Image.open(BytesIO(data)) - new_file = BytesIO() - # Determine if the image is a GIF - is_animated = getattr(image, "is_animated", False) - if is_animated: - frames = [frame.convert("RGBA") for frame in ImageSequence.Iterator(image)] - # Save the new GIF - frames[0].save( - new_file, - format='GIF', - save_all=True, - append_images=frames[1:], - loop=image.info.get('loop', 0), # Default loop to 0 if not present - duration=image.info.get('duration', 100), # Set a default duration if not present - disposal=image.info.get('disposal', 2) # Default to disposal method 2 (restore to background) - ) - # Get the size of the first frame to determine resizing - w, h = frames[0].size - else: - suffix = mimetypes.guess_extension(mimetype) - if suffix: - suffix = suffix[1:] - image = image.convert("RGBA") - image.save(new_file, format=suffix) - w, h = image.size - if w > 256 or h > 256: - # Set the width and height to lower values so clients wouldn't show them as huge images - if w > h: - h = int(h / (w / 256)) - w = 256 - else: - w = int(w / (h / 256)) - h = 256 - return new_file.getvalue(), w, h + with Image.open(BytesIO(data)) as image: + with BytesIO() as new_file: + # Determine if the image is a GIF + is_animated = getattr(image, "is_animated", False) + if is_animated: + frames = [frame.convert("RGBA") for frame in ImageSequence.Iterator(image)] + # Save the new GIF + frames[0].save( + new_file, + format='GIF', + save_all=True, + append_images=frames[1:], + loop=image.info.get('loop', 0), # Default loop to 0 if not present + duration=image.info.get('duration', 100), # Set a default duration if not present + disposal=image.info.get('disposal', 2) # Default to disposal method 2 (restore to background) + ) + # Get the size of the first frame to determine resizing + w, h = frames[0].size + else: + suffix = mimetypes.guess_extension(mimetype) + if suffix: + suffix = suffix[1:] + image = image.convert("RGBA") + image.save(new_file, format=suffix) + w, h = image.size + if w > 256 or h > 256: + # Set the width and height to lower values so clients wouldn't show them as huge images + if w > h: + h = int(h / (w / 256)) + w = 256 + else: + w = int(w / (h / 256)) + h = 256 + return new_file.getvalue(), w, h def _convert_sticker(data: bytes) -> (bytes, str, int, int):