Merge branch 'master' into enhancement/typst-binary

Dockerfile

@@ -1,16 +1,3 @@
-FROM alpine:latest AS artifacts
-
-RUN apk add --no-cache curl xz
-WORKDIR /tmp
-
-RUN mkdir -p /artifacts
-RUN curl -L -o typst.tar.xz "https://github.com/typst/typst/releases/download/v0.14.2/typst-x86_64-unknown-linux-musl.tar.xz" \
-    && tar -xJf typst.tar.xz \
-    && mv typst-x86_64-unknown-linux-musl/typst /artifacts
-
-RUN chmod -R +x /artifacts/
-
-
 FROM python:3.13-slim AS base
 
 ENV VIRTUAL_ENV=/app/.venv \
@@ -51,7 +38,6 @@ RUN uv sync --no-install-project
 FROM base AS runtime
 
 COPY --from=builder ${VIRTUAL_ENV} ${VIRTUAL_ENV}
-COPY --from=artifacts /artifacts/ /usr/local/bin/
 
 WORKDIR /app
 

konabot/common/artifact.py

@@ -1,9 +1,10 @@
 import asyncio
+from typing import Any, Awaitable, Callable
 import aiohttp
 import hashlib
 import platform
 
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from pathlib import Path
 
 import nonebot
@@ -14,6 +15,8 @@ from pydantic import BaseModel
 
 @dataclass
 class ArtifactDepends:
+    _Callback = Callable[[bool], Awaitable[Any]]
+
     url: str
     sha256: str
     target: Path
@@ -27,6 +30,9 @@ class ArtifactDepends:
     use_proxy: bool = True
     "网络问题，赫赫；使用的是 Discord 模块配置的 proxy"
 
+    callbacks: list[_Callback] = field(default_factory=list)
+    "在任务完成以后，应该做的事情"
+
     def is_corresponding_platform(self) -> bool:
         if self.required_os is not None:
             if self.required_os.lower() != platform.system().lower():
@@ -36,26 +42,43 @@ class ArtifactDepends:
                 return False
         return True
 
+    def on_finished(self, task: _Callback) -> _Callback:
+        self.callbacks.append(task)
+        return task
+
+    async def _finished(self, downloaded: bool) -> list[Any | BaseException]:
+        tasks = set()
+        for f in self.callbacks:
+            tasks.add(f(downloaded))
+        return await asyncio.gather(*tasks, return_exceptions=True)
+
 
 class Config(BaseModel):
     prefetch_artifact: bool = False
     "是否提前下载好二进制依赖"
 
 
-artifact_list = []
+artifact_list: list[ArtifactDepends] = []
 
 
 driver = nonebot.get_driver()
 config = nonebot.get_plugin_config(Config)
 
+
 @driver.on_startup
 async def _():
     if config.prefetch_artifact:
         logger.info("启动检测中：正在检测需求的二进制是否下载")
         semaphore = asyncio.Semaphore(10)
+
         async def _task(artifact: ArtifactDepends):
             async with semaphore:
-                await ensure_artifact(artifact)
+                downloaded = await ensure_artifact(artifact)
+                result = await artifact._finished(downloaded)
+                for r in result:
+                    if isinstance(r, BaseException):
+                        logger.warning("完成了二进制文件的下载，但是有未捕捉的错误")
+                        logger.exception(r)
 
         tasks: set[asyncio.Task] = set()
         for a in artifact_list:
@@ -78,35 +101,43 @@ async def download_artifact(artifact: ArtifactDepends):
     async with aiohttp.ClientSession(proxy=proxy) as client:
         result = await client.get(artifact.url)
         if result.status != 200:
-            logger.warning(f"已经下载了二进制，但是注意服务器没有返回 200！ URL={artifact.url} TARGET={artifact.target} CODE={result.status}")
+            logger.warning(
+                f"已经下载了二进制，但是注意服务器没有返回 200！ URL={artifact.url} TARGET={artifact.target} CODE={result.status}"
+            )
         data = await result.read()
     artifact.target.write_bytes(data)
-    if not platform.system().lower() == 'windows':
+    if not platform.system().lower() == "windows":
         artifact.target.chmod(0o755)
 
     logger.info(f"下载好了 TARGET={artifact.target} URL={artifact.url}")
     m = hashlib.sha256(artifact.target.read_bytes())
     if m.hexdigest().lower() != artifact.sha256.lower():
-        logger.warning(f"下载到的二进制的 sha256 与需求不同 TARGET={artifact.target} REQUESTED={artifact.sha256} ACTUAL={m.hexdigest()}")
+        logger.warning(
+            f"下载到的二进制的 sha256 与需求不同 TARGET={artifact.target} REQUESTED={artifact.sha256} ACTUAL={m.hexdigest()}"
+        )
 
 
-async def ensure_artifact(artifact: ArtifactDepends):
+async def ensure_artifact(artifact: ArtifactDepends) -> bool:
     if not artifact.is_corresponding_platform():
-        return
+        return False
 
     if not artifact.target.exists():
         logger.info(f"二进制依赖 {artifact.target} 不存在")
         if not artifact.target.parent.exists():
             artifact.target.parent.mkdir(parents=True, exist_ok=True)
         await download_artifact(artifact)
+        return True
     else:
         m = hashlib.sha256(artifact.target.read_bytes())
         if m.hexdigest().lower() != artifact.sha256.lower():
-            logger.info(f"二进制依赖 {artifact.target} 的哈希无法对应需求的哈希，准备重新下载")
+            logger.info(
+                f"二进制依赖 {artifact.target} 的哈希无法对应需求的哈希，准备重新下载"
+            )
             artifact.target.unlink()
             await download_artifact(artifact)
+            return True
+        return False
 
 
 def register_artifacts(*artifacts: ArtifactDepends):
     artifact_list.extend(artifacts)
-

konabot/docs/user/fx.txt

@@ -79,14 +79,6 @@ fx [滤镜名称] <参数1> <参数2> ...
 * ```fx JPEG损坏 <质量=10>```
   * 质量范围建议为 1~95，数值越低，压缩痕迹越重、效果越搞笑。
 * ```fx 动图 <帧率=10>```
-* ```fx 像素排序 <方向=horizontal> <阈值=0> <自动阈值=true> <排序依据=brightness> <遮罩阈值=128> <反向=false> <块大小=1>```
-  * 对像素按指定属性进行排序，效果类似 Photoshop/GIMP Pixel Sort。
-  * **方向**：horizontal（水平）/ vertical（垂直）
-  * **排序依据**：brightness（亮度）/ hue（色相）/ red / green / blue
-  * **自动阈值**：true 时使用图像亮度中位数作为遮罩阈值
-  * **遮罩阈值**：决定哪些像素参与排序（亮度 >= 阈值）
-  * **反向**：true 时从亮到暗排序
-  * **块大小**：每 N 行/列作为一个整体排序单位
 
 ### 多图像处理器
 * ```fx 存入图像 <目标名称>```

konabot/docs/user/giftool.txt

@@ -71,14 +71,6 @@ giftool [图片] [选项]
 
 - 调整 GIF 图的速度。若为负数，则代表倒放。
 
-### `--pingpong`（可选）
-
-- 开启乒乓模式，生成正放-倒放拼接的 GIF 图。
-- 即播放完正向后，会倒放回去，形成往复循环效果。
-- 可与 `--speed` 配合使用，调整播放速度。
-- 示例：`giftool [图片] --pingpong`
-- 示例：`giftool [图片] --pingpong --speed 2.0`
-
 ## 使用方式
 
 1. 发送指令前，请确保：

konabot/plugins/fx_process/fx_handle.py

@@ -1354,140 +1354,6 @@ class ImageFilterImplement:
         images.append(text_image)
         return image
 
-    # Pixel Sort - 像素排序效果
-    @staticmethod
-    def apply_pixel_sort(
-        image: Image.Image,
-        direction: str = "horizontal",
-        threshold: float = 0.0,
-        auto_threshold: bool = True,
-        sort_by: str = "brightness",
-        mask_threshold: float = 128.0,
-        reverse: bool = False,
-        block_size: int = 1
-    ) -> Image.Image:
-        """
-        Pixel Sort 效果
-        
-        参数:
-            image: 输入图像
-            direction: 排序方向，"horizontal"(水平) 或 "vertical"(垂直)
-            threshold: 亮度阈值 (0-255)，低于此值的像素会被排序（仅在 auto_threshold=False 时生效）
-            auto_threshold: 是否自动计算阈值（使用图像中位数）
-            sort_by: 排序依据，"brightness"(亮度)、"hue"(色相)、"red"、"green"、"blue"
-            mask_threshold: 遮罩阈值 (0-255)，决定哪些像素参与排序
-            reverse: 是否反向排序
-            block_size: 块大小，每 N 行/列作为一个整体排序单位
-        """
-        if image.mode != 'RGBA':
-            image = image.convert('RGBA')
-        
-        arr = np.array(image)
-        height, width = arr.shape[:2]
-        
-        # 获取排序属性
-        def get_sort_value(pixel):
-            r, g, b = pixel[0], pixel[1], pixel[2]
-            if sort_by == "brightness":
-                return 0.299 * r + 0.587 * g + 0.114 * b
-            elif sort_by == "hue":
-                max_c = max(r, g, b)
-                min_c = min(r, g, b)
-                diff = max_c - min_c
-                if diff == 0:
-                    return 0
-                if max_c == r:
-                    return 60 * (((g - b) / diff) % 6)
-                elif max_c == g:
-                    return 60 * ((b - r) / diff + 2)
-                else:
-                    return 60 * ((r - g) / diff + 4)
-            elif sort_by == "red":
-                return r
-            elif sort_by == "green":
-                return g
-            elif sort_by == "blue":
-                return b
-            return 0.299 * r + 0.587 * g + 0.114 * b
-        
-        # 自动计算阈值
-        if auto_threshold:
-            # 使用图像亮度中位数作为阈值
-            gray = np.array(image.convert('L'))
-            mask_threshold = float(np.median(gray))
-        
-        # 创建遮罩：哪些像素需要排序
-        mask = np.zeros((height, width), dtype=bool)
-        for y in range(height):
-            for x in range(width):
-                brightness = 0.299 * arr[y, x, 0] + 0.587 * arr[y, x, 1] + 0.114 * arr[y, x, 2]
-                mask[y, x] = brightness >= mask_threshold
-        
-        result = arr.copy()
-        
-        if direction.lower() in ["horizontal", "h", "水平"]:
-            # 水平排序（逐行）
-            for y in range(height):
-                # 收集当前行中需要排序的像素
-                if block_size > 1:
-                    # 按块处理
-                    for block_start in range(0, width, block_size):
-                        block_end = min(block_start + block_size, width)
-                        pixels = []
-                        indices = []
-                        for x in range(block_start, block_end):
-                            if mask[y, x]:
-                                pixels.append(arr[y, x].copy())
-                                indices.append(x)
-                        if len(pixels) > 1:
-                            # 按指定属性排序
-                            sorted_pixels = sorted(pixels, key=get_sort_value, reverse=reverse)
-                            for i, x in enumerate(indices):
-                                result[y, x] = sorted_pixels[i]
-                else:
-                    # 逐像素处理
-                    pixels = []
-                    indices = []
-                    for x in range(width):
-                        if mask[y, x]:
-                            pixels.append(arr[y, x].copy())
-                            indices.append(x)
-                    if len(pixels) > 1:
-                        sorted_pixels = sorted(pixels, key=get_sort_value, reverse=reverse)
-                        for i, x in enumerate(indices):
-                            result[y, x] = sorted_pixels[i]
-        
-        elif direction.lower() in ["vertical", "v", "垂直"]:
-            # 垂直排序（逐列）
-            for x in range(width):
-                if block_size > 1:
-                    # 按块处理
-                    for block_start in range(0, height, block_size):
-                        block_end = min(block_start + block_size, height)
-                        pixels = []
-                        indices = []
-                        for y in range(block_start, block_end):
-                            if mask[y, x]:
-                                pixels.append(arr[y, x].copy())
-                                indices.append(y)
-                        if len(pixels) > 1:
-                            sorted_pixels = sorted(pixels, key=get_sort_value, reverse=reverse)
-                            for i, y in enumerate(indices):
-                                result[y, x] = sorted_pixels[i]
-                else:
-                    pixels = []
-                    indices = []
-                    for y in range(height):
-                        if mask[y, x]:
-                            pixels.append(arr[y, x].copy())
-                            indices.append(y)
-                    if len(pixels) > 1:
-                        sorted_pixels = sorted(pixels, key=get_sort_value, reverse=reverse)
-                        for i, y in enumerate(indices):
-                            result[y, x] = sorted_pixels[i]
-        
-        return Image.fromarray(result, 'RGBA')
-
 
     
 class ImageFilterEmpty:

konabot/plugins/fx_process/fx_manager.py

@@ -65,8 +65,6 @@ class ImageFilterManager:
         "覆盖图像": ImageFilterImplement.apply_overlay,
         # 生成式
         "覆加颜色": ImageFilterImplement.generate_solid,
-        # Pixel Sort
-        "像素排序": ImageFilterImplement.apply_pixel_sort,
     }
 
     generate_filter_map = {

konabot/plugins/image_process/__init__.py

@@ -6,7 +6,7 @@ import PIL
 import PIL.Image
 import cv2
 import imageio.v3 as iio
-from nonebot_plugin_alconna import Alconna, Args, Image, Option, Query, UniMessage, on_alconna
+from nonebot_plugin_alconna import Alconna, Args, Image, Option, UniMessage, on_alconna
 import numpy
 
 from konabot.common.nb.exc import BotExceptionMessage
@@ -34,7 +34,6 @@ cmd_giftool = on_alconna(
         Option("-t", Args["length", str]),
         Option("-to", Args["end_point", str]),
         Option("--speed", Args["speed_factor", float], default=1.0, alias=["-s"]),
-        Option("--pingpong"),
     )
 )
 
@@ -47,7 +46,6 @@ async def _(
     length: str | None = None,
     speed_factor: float = 1.0,
     end_point: str | None = None,
-    pingpong: Query[bool] = Query("pingpong"),
 ):
     ss: None | float = None
     if start_point:
@@ -164,16 +162,6 @@ async def _(
         rframes = rframes[::-1]
         rdur_ms = rdur_ms[::-1]
 
-    # 处理 pingpong 模式
-    if pingpong.available:
-        # 复制一份反转的帧序列（去掉第一帧避免重复）
-        pingpong_frames = rframes[1:][::-1] if len(rframes) > 1 else rframes[::-1]
-        pingpong_durations = rdur_ms[1:][::-1] if len(rdur_ms) > 1 else rdur_ms[::-1]
-        
-        # 拼接正放和倒放
-        rframes = rframes + pingpong_frames
-        rdur_ms = rdur_ms + pingpong_durations
-
     output_img = BytesIO()
 
     if rframes:

konabot/plugins/memepack/__init__.py

@@ -34,7 +34,6 @@ from konabot.plugins.memepack.drawing.saying import (
     draw_mnk,
     draw_pt,
     draw_suan,
-    draw_vr,
 )
 from konabot.plugins.memepack.drawing.watermark import draw_doubao_watermark
 
@@ -335,29 +334,3 @@ async def _(img: DepPILImage):
     result_bytes = BytesIO()
     result.save(result_bytes, format="PNG")
     await doubao_cmd.send(await UniMessage().image(raw=result_bytes).export())
-
-
-vrsay = on_alconna(
-    Alconna(
-        "vr说",
-        Args[
-            "saying",
-            MultiVar(str, "+"),
-            Field(missing_tips=lambda: "你没有写vr说了什么"),
-        ],
-    ),
-    use_cmd_start=True,
-    use_cmd_sep=False,
-    skip_for_unmatch=False,
-    aliases=set(),
-)
-
-
-@vrsay.handle()
-async def _(saying: list[str]):
-    img = await draw_vr("\n".join(saying))
-    img_bytes = BytesIO()
-    img.save(img_bytes, format="PNG")
-
-    await vrsay.send(await UniMessage().image(raw=img_bytes).export())
-

konabot/plugins/memepack/drawing/saying.py

@@ -16,7 +16,6 @@ dasuan_image = PIL.Image.open(ASSETS_PATH / "img" / "meme" / "dss.png").convert(
 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")
 kio_image = PIL.Image.open(ASSETS_PATH / "img" / "meme" / "kiosay.jpg").convert("RGBA")
-vr_image = PIL.Image.open(ASSETS_PATH / 'img' / 'meme' / 'vr.jpg').convert("RGBA")
 
 
 def _draw_geimao(saying: str):
@@ -124,24 +123,3 @@ def draw_kiosay(saying: str):
         )
     return img
 
-
-@make_async
-def draw_vr(saying: str):
-    img = vr_image.copy()
-    w, h = img.size
-    hw = 300
-
-    img2 = PIL.Image.new("RGBA", (w, h + hw), 'white')
-    img2.paste(img, (0, hw))
-
-    with imagetext_py.Writer(img2) as iw:
-        iw.draw_text_wrapped(
-            saying, w // 2, hw // 2 + 15, 0.5, 0.5, w, 64, LXGWWENKAI_REGULAR,
-            imagetext_py.Paint.Color(imagetext_py.Color.from_hex("000000FF")),
-            1.0,
-            imagetext_py.TextAlign.Center,
-            draw_emojis=True,
-        )
-
-    return img2
-

konabot/plugins/typst/__init__.py

@@ -1,9 +1,11 @@
 import asyncio
+import os
 import subprocess
 
 from pathlib import Path
 from tempfile import TemporaryDirectory
 from typing import cast
+import zipfile
 
 from loguru import logger
 from nonebot import on_command
@@ -13,22 +15,99 @@ from nonebot.adapters.onebot.v11.event import MessageEvent as OB11MessageEvent
 from nonebot.adapters.onebot.v11.bot import Bot as OB11Bot
 from nonebot.adapters.onebot.v11.message import Message as OB11Message
 
+from konabot.common.artifact import ArtifactDepends, ensure_artifact, register_artifacts
 from konabot.common.longtask import DepLongTaskTarget
-from konabot.common.path import TMP_PATH
+from konabot.common.path import BINARY_PATH, TMP_PATH
+
+
+arti_typst_linux = ArtifactDepends(
+    url="https://github.com/typst/typst/releases/download/v0.14.2/typst-x86_64-unknown-linux-musl.tar.xz",
+    sha256="a6044cbad2a954deb921167e257e120ac0a16b20339ec01121194ff9d394996d",
+    target=BINARY_PATH / "typst.tar.xz",
+    required_os="Linux",
+    required_arch="x86_64",
+)
+arti_typst_windows = ArtifactDepends(
+    url="https://github.com/typst/typst/releases/download/v0.14.2/typst-x86_64-pc-windows-msvc.zip",
+    sha256="51353994ac83218c3497052e89b2c432c53b9d4439cdc1b361e2ea4798ebfc13",
+    target=BINARY_PATH / "typst.zip",
+    required_os="Windows",
+    required_arch="AMD64",
+)
+
+
+bin_path: Path | None = None
+
+
+@arti_typst_linux.on_finished
+async def _(downloaded: bool):
+    global bin_path
+
+    tar_path = arti_typst_linux.target
+    bin_path = BINARY_PATH / "typst"
+
+    if downloaded or not bin_path.exists():
+        bin_path.unlink(missing_ok=True)
+        process = await asyncio.create_subprocess_exec(
+            "tar",
+            "-xvf",
+            tar_path,
+            "--strip-components=1",
+            "-C",
+            BINARY_PATH,
+            "typst-x86_64-unknown-linux-musl/typst",
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+        )
+        stdout, stderr = await process.communicate()
+        if process.returncode != 0 or not bin_path.exists():
+            logger.warning(
+                "似乎没有成功解压 Typst 二进制文件，检查一下吧！ "
+                f"stdout={stdout} stderr={stderr}"
+            )
+        else:
+            os.chmod(bin_path, 0o755)
+
+
+@arti_typst_windows.on_finished
+async def _(downloaded: bool):
+    global bin_path
+    zip_path = arti_typst_windows.target
+    bin_path = BINARY_PATH / "typst.exe"
+
+    if downloaded or not bin_path.exists():
+        bin_path.unlink(missing_ok=True)
+        with zipfile.ZipFile(zip_path, "r") as zf:
+            target_name = "typst-x86_64-pc-windows-msvc/typst.exe"
+            if target_name not in zf.namelist():
+                logger.warning("在 Zip 压缩包里面没有找到目标文件")
+                return
+            zf.extract(target_name, BINARY_PATH)
+            (BINARY_PATH / target_name).rename(bin_path)
+            (BINARY_PATH / "typst-x86_64-pc-windows-msvc").rmdir()
+
+
+register_artifacts(arti_typst_linux)
+register_artifacts(arti_typst_windows)
 
 
 TEMPLATE_PATH = Path(__file__).parent / "template.typ"
 TEMPLATE = TEMPLATE_PATH.read_text()
 
 
-def render_sync(code: str) -> bytes:
+def render_sync(code: str) -> bytes | None:
+    global bin_path
+
+    if bin_path is None:
+        return
+
     with TemporaryDirectory(dir=TMP_PATH) as tmpdirname:
         temp_dir = Path(tmpdirname).resolve()
         temp_typ = temp_dir / "page.typ"
         temp_typ.write_text(TEMPLATE + "\n\n" + code)
 
         cmd = [
-            "typst",
+            bin_path,
             "compile",
             temp_typ.name,
             "--format",
@@ -61,7 +140,7 @@ def render_sync(code: str) -> bytes:
         return result_png.read_bytes()
 
 
-async def render(code: str) -> bytes:
+async def render(code: str) -> bytes | None:
     task = asyncio.to_thread(lambda: render_sync(code))
     return await task
 
@@ -70,7 +149,21 @@ cmd = on_command("typst")
 
 
 @cmd.handle()
-async def _(evt: Event, bot: Bot, msg: UniMsg, target: DepLongTaskTarget):
+async def _(
+    evt: Event,
+    bot: Bot,
+    msg: UniMsg,
+    target: DepLongTaskTarget,
+):
+    global bin_path
+
+    # 对于本地机器，一般不会在应用启动时自动下载，这里再保证存在
+    await ensure_artifact(arti_typst_linux)
+    await ensure_artifact(arti_typst_windows)
+    if bin_path is None or not bin_path.exists():
+        logger.warning("当前环境不存在 Typst，但仍然调用了")
+        return
+
     typst_code = ""
     if isinstance(evt, OB11MessageEvent):
         if evt.reply is not None:
@@ -92,6 +185,8 @@ async def _(evt: Event, bot: Bot, msg: UniMsg, target: DepLongTaskTarget):
 
     try:
         res = await render(typst_code)
+        if res is None:
+            raise FileNotFoundError("没有渲染出来内容")
     except FileNotFoundError as e:
         await target.send_message("渲染出错：内部错误")
         raise e from e

tests/test_fx_process.py

@@ -86,67 +86,3 @@ def test_prase_input_args_parses_resize_second_argument_as_float():
     assert len(filters) == 1
     assert filters[0].name == "缩放"
     assert filters[0].args == [2.0, 3.0]
-
-
-def test_apply_pixel_sort_keeps_image_mode_and_size():
-    """测试 Pixel Sort 保持图像的 mode 和 size"""
-    image = Image.new("RGBA", (10, 10), (255, 0, 0, 128))
-
-    result = ImageFilterImplement.apply_pixel_sort(image)
-
-    assert result.size == image.size
-    assert result.mode == "RGBA"
-
-
-def test_apply_pixel_sort_horizontal():
-    """测试水平方向的 Pixel Sort"""
-    # 创建一个简单的渐变图像
-    image = Image.new("RGB", (5, 3))
-    # 第一行：红到蓝渐变
-    image.putpixel((0, 0), (255, 0, 0))
-    image.putpixel((1, 0), (200, 0, 0))
-    image.putpixel((2, 0), (100, 0, 0))
-    image.putpixel((3, 0), (50, 0, 0))
-    image.putpixel((4, 0), (0, 0, 255))
-    # 填充其他行
-    for y in range(1, 3):
-        for x in range(5):
-            image.putpixel((x, y), (128, 128, 128))
-
-    result = ImageFilterImplement.apply_pixel_sort(
-        image, direction="horizontal", auto_threshold=False, mask_threshold=10
-    )
-
-    assert result.size == image.size
-    assert result.mode == "RGBA"
-
-
-def test_apply_pixel_sort_vertical():
-    """测试垂直方向的 Pixel Sort"""
-    image = Image.new("RGB", (3, 5))
-    # 第一列：绿到红渐变
-    image.putpixel((0, 0), (0, 255, 0))
-    image.putpixel((0, 1), (0, 200, 0))
-    image.putpixel((0, 2), (0, 100, 0))
-    image.putpixel((0, 3), (0, 50, 0))
-    image.putpixel((0, 4), (255, 0, 0))
-    # 填充其他列
-    for y in range(5):
-        for x in range(1, 3):
-            image.putpixel((x, y), (128, 128, 128))
-
-    result = ImageFilterImplement.apply_pixel_sort(
-        image, direction="vertical", auto_threshold=False, mask_threshold=10
-    )
-
-    assert result.size == image.size
-    assert result.mode == "RGBA"
-
-
-def test_prase_input_args_parses_pixel_sort_arguments():
-    """测试解析 Pixel Sort 参数"""
-    filters = prase_input_args("像素排序 horizontal 0 false brightness 128 false 1")
-
-    assert len(filters) == 1
-    assert filters[0].name == "像素排序"
-    assert filters[0].args == ["horizontal", 0.0, False, "brightness", 128.0, False, 1]