接入我写的模块来获得群上下文

Reviewed-on: #26

.drone.yml

@@ -10,6 +10,10 @@ trigger:
     - master
 
 steps:
+  - name: submodules
+    image: alpine/git
+    commands:
+    - git submodule update --init --recursive
   - name: 构建 Docker 镜像
     image: plugins/docker:latest
     privileged: true
@@ -50,6 +54,10 @@ trigger:
     - tag
 
 steps:
+  - name: submodules
+    image: alpine/git
+    commands:
+    - git submodule update --init --recursive
   - name: 构建并推送 Release Docker 镜像
     image: plugins/docker:latest
     privileged: true

.gitmodules

@@ -0,0 +1,3 @@
+[submodule "assets/lexicon/THUOCL"]
+	path = assets/lexicon/THUOCL
+	url = https://github.com/thunlp/THUOCL.git

Dockerfile

@@ -4,6 +4,15 @@ FROM python:3.13-slim AS base
 ENV VIRTUAL_ENV=/app/.venv \
     PATH="/app/.venv/bin:$PATH"
 
+RUN apt-get update && \
+    apt-get install -y --no-install-recommends \
+    libfontconfig1 \
+    libgl1 \
+    libegl1 \
+    libglvnd0 \
+    mesa-vulkan-drivers \
+    && rm -rf /var/lib/apt/lists/*
+
 
 
 FROM base AS builder

assets/json/poll.json

@@ -0,0 +1 @@
+{"poll": {"0": {"create": 1760357553, "expiry": 1760443953, "options": {"0": "此方bot", "1": "testpilot", "2": "小镜bot", "3": "可怜bot"}, "polldata": {}, "qq": "2975499623", "title": "我~是~谁~？"}}}

bot.py

@@ -7,6 +7,10 @@ from nonebot.adapters.discord import Adapter as DiscordAdapter
 from nonebot.adapters.minecraft import Adapter as MinecraftAdapter
 from nonebot.adapters.onebot.v11 import Adapter as OnebotAdapter
 
+from konabot.common.log import init_logger
+from konabot.common.nb.exc import BotExceptionMessage
+from konabot.common.path import LOG_PATH
+
 dotenv.load_dotenv()
 env = os.environ.get("ENVIRONMENT", "prod")
 env_enable_console = os.environ.get("ENABLE_CONSOLE", "none")
@@ -14,7 +18,12 @@ env_enable_qq = os.environ.get("ENABLE_QQ", "none")
 env_enable_discord = os.environ.get("ENABLE_DISCORD", "none")
 env_enable_minecraft = os.environ.get("ENABLE_MINECRAFT", "none")
 
+
 def main():
+    init_logger(LOG_PATH, [
+        BotExceptionMessage,
+    ])
+
     nonebot.init()
 
     driver = nonebot.get_driver()
@@ -33,6 +42,7 @@ def main():
 
     # nonebot.load_builtin_plugin("echo")
     nonebot.load_plugins("konabot/plugins")
+    nonebot.load_plugin("nonebot_plugin_analysis_bilibili")
 
     nonebot.run()
 

konabot/common/data_man.py

@@ -0,0 +1,36 @@
+import asyncio
+from contextlib import asynccontextmanager
+from pathlib import Path
+from typing import Generic, TypeVar
+
+from pydantic import BaseModel, ValidationError
+
+T = TypeVar("T", bound=BaseModel)
+
+
+class DataManager(Generic[T]):
+    def __init__(self, cls: type[T], fp: Path) -> None:
+        self.cls = cls
+        self.fp = fp
+        self._aio_lock = asyncio.Lock()
+        self._data: T | None = None
+
+    def load(self) -> T:
+        if not self.fp.exists():
+            return self.cls()
+        try:
+            return self.cls.model_validate_json(self.fp.read_text())
+        except ValidationError:
+            return self.cls()
+
+    def save(self, data: T):
+        self.fp.write_text(data.model_dump_json())
+
+    @asynccontextmanager
+    async def get_data(self):
+        await self._aio_lock.acquire()
+        self._data = self.load()
+        yield self._data
+        self.save(self._data)
+        self._data = None
+        self._aio_lock.release()

konabot/common/log.py

@@ -0,0 +1,79 @@
+import sys
+from pathlib import Path
+from typing import TYPE_CHECKING, List, Type
+
+from loguru import logger
+
+if TYPE_CHECKING:
+    from loguru import Record
+
+
+def file_exception_filter(
+    record: "Record", 
+    ignored_exceptions: tuple[Type[Exception], ...]
+) -> bool:
+    """
+    一个自定义的 Loguru 过滤器函数。
+    如果日志记录包含异常信息，并且该异常的类型在 ignored_exceptions 中，则返回 False（忽略）。
+    否则，返回 True（允许记录）。
+    """
+    exception_info = record.get("exception")
+    
+    if exception_info:
+        exception_type = exception_info[0]
+
+        if exception_type and issubclass(exception_type, ignored_exceptions):
+            return False
+
+    return True
+
+
+def init_logger(
+    log_dir: Path, 
+    ignored_exceptions: List[Type[Exception]]
+) -> None:
+    """
+    配置全局 Loguru Logger。
+
+    Args:
+        log_dir (Path): 存放日志文件的文件夹路径，会自动创建。
+        ignored_exceptions (List[Type[Exception]]): 在 WARNING 级别文件日志中需要忽略的异常类型列表。
+    """
+
+    ignored_exceptions_tuple = tuple(ignored_exceptions)
+    logger.remove()
+
+    log_dir.mkdir(parents=True, exist_ok=True)
+
+    logger.add(
+        sys.stderr,
+        level="INFO",
+        colorize=True,
+        format="<green>{time:HH:mm:ss}</green> | <level>{level: <8}</level> | <cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - <level>{message}</level>",
+    )
+
+    info_log_path = log_dir / "log.log"
+    logger.add(
+        str(info_log_path),
+        level="INFO",
+        rotation="10 MB",
+        retention="7 days",
+        enqueue=True,
+        backtrace=False,
+        diagnose=False,
+    )
+
+    warning_error_log_path = log_dir / "error.log"
+    logger.add(
+        str(warning_error_log_path),
+        level="WARNING",
+        rotation="10 MB",
+        compression="zip",
+        enqueue=True,
+        filter=lambda record: file_exception_filter(record, ignored_exceptions_tuple),
+        backtrace=True,
+        diagnose=True,
+    )
+
+    logger.info("Loguru Logger 初始化完成！")
+    logger.info(f"控制台日志级别: INFO")

konabot/common/longtask.py

@@ -0,0 +1,275 @@
+from contextlib import asynccontextmanager
+import datetime
+import json
+from typing import Annotated, Any, Callable, Coroutine, cast
+import asyncio as asynkio
+import uuid
+
+from loguru import logger
+import nonebot
+from nonebot.params import Depends
+from nonebot.adapters import Event as BaseEvent
+from nonebot.adapters import Bot as BaseBot
+from nonebot.adapters.onebot.v11 import Bot as OBBot
+from nonebot.adapters.onebot.v11 import GroupMessageEvent as OBGroupMessageEvent
+from nonebot.adapters.onebot.v11 import PrivateMessageEvent as OBPrivateMessageEvent
+from nonebot.adapters.console import Bot as ConsoleBot
+from nonebot.adapters.console import MessageEvent as ConsoleMessageEvent
+from nonebot.adapters.discord import MessageEvent as DCMessageEvent
+from nonebot.adapters.discord import Bot as DCBot
+from nonebot_plugin_alconna import UniMessage
+from pydantic import BaseModel, ValidationError
+
+from .path import DATA_PATH
+
+LONGTASK_DATA_DIR = DATA_PATH / "longtasks.json"
+
+
+class LongTaskTarget(BaseModel):
+    """
+    用于定义长期任务的目标沟通对象，一般通过 DepLongTaskTarget 依赖注入获取：
+
+    ```python
+    @cmd.handle()
+    async def _(target: DepLongTaskTarget):
+        ...
+    ```
+    """
+
+    platform: str
+    "沟通对象所在的平台"
+
+    self_id: str
+    "进行沟通的对象自己的 ID"
+
+    channel_id: str
+    "沟通对象所在的群或者 Discord Channel。若为空则代表是私聊"
+
+    target_id: str
+    "沟通对象的 ID"
+
+    async def send_message(self, msg: UniMessage, at: bool = True) -> bool:
+        try:
+            bot = nonebot.get_bot(self.self_id)
+        except KeyError:
+            logger.warning(f"试图访问了不存在的 Bot。ID={self.self_id}")
+            return False
+
+        if self.platform == "qq":
+            if not isinstance(bot, OBBot):
+                logger.warning(
+                    f"编号对应的平台并非期望的平台 ID={self.self_id} PLATFORM={
+                        self.platform
+                    } BOT_CLASS={bot.__class__.__name__}"
+                )
+                return False
+            if self.channel_id == "":
+                # 私聊模式
+                await bot.send_private_msg(
+                    user_id=int(self.target_id),
+                    message=cast(Any, await msg.export(bot)),
+                    auto_escape=False,
+                )
+                return True
+            else:
+                if at:
+                    msg = UniMessage().at(self.target_id).text(" ") + msg
+                await bot.send_group_msg(
+                    group_id=int(self.channel_id),
+                    message=cast(Any, await msg.export(bot)),
+                    auto_escape=False,
+                )
+                return True
+        if self.platform == "console":
+            if not isinstance(bot, ConsoleBot):
+                logger.warning(
+                    f"编号对应的平台并非期望的平台 ID={self.self_id} PLATFORM={
+                        self.platform
+                    } BOT_CLASS={bot.__class__.__name__}"
+                )
+                return False
+            await bot.send_message(self.channel_id, cast(Any, await msg.export()))
+        if self.platform == "discord":
+            if not isinstance(bot, DCBot):
+                logger.warning(
+                    f"编号对应的平台并非期望的平台 ID={self.self_id} PLATFORM={
+                        self.platform
+                    } BOT_CLASS={bot.__class__.__name__}"
+                )
+                return False
+            await bot.send_to(
+                channel_id=int(self.channel_id),
+                message=cast(
+                    Any, await (UniMessage().at(self.target_id) + msg).export()
+                ),
+                tts=False,
+            )
+        logger.warning(f"没有一个平台是期望的平台 PLATFORM={self.platform}")
+        return False
+
+
+class LongTask(BaseModel):
+    uuid: str
+    data_json: str
+    target: "LongTaskTarget"
+    callback: str
+    deadline: datetime.datetime
+    canceled: bool = False
+
+    _aio_task: asynkio.Task | None = None
+
+    async def run(self):
+        now = datetime.datetime.now()
+        if self.deadline < now and not self.canceled:
+            await self._run_task()
+            return
+        await asynkio.sleep((self.deadline - now).total_seconds())
+        if self.canceled:
+            return
+        await self._run_task()
+
+    async def _run_task(self):
+        hdl = registered_long_task_handler.get(self.callback, None)
+        if hdl is None:
+            logger.warning(
+                f"Callback {self.callback} 未曾被注册，但是被期待调用，已忽略"
+            )
+            async with longtask_data() as datafile:
+                datafile.to_handle[self.callback] = [
+                    t
+                    for t in datafile.to_handle.get(self.callback, [])
+                    if t.uuid != self.uuid
+                ]
+                datafile.unhandled.setdefault(self.callback, []).append(self)
+
+            return
+        await hdl(self)
+        async with longtask_data() as datafile:
+            datafile.to_handle[self.callback] = [
+                t for t in datafile.to_handle[self.callback] if t.uuid != self.uuid
+            ]
+
+    def clean(self):
+        self._aio_task = None
+
+    @property
+    def data(self):
+        return json.loads(self.data_json)
+
+    async def start(self):
+        self._aio_task = asynkio.Task(self.run())
+        self._aio_task.add_done_callback(lambda _: self.clean())
+
+
+class LongTaskModuleData(BaseModel):
+    to_handle: dict[str, list[LongTask]]
+    unhandled: dict[str, list[LongTask]]
+
+
+async def get_long_task_target(event: BaseEvent, bot: BaseBot) -> LongTaskTarget | None:
+    if isinstance(event, OBGroupMessageEvent):
+        return LongTaskTarget(
+            platform="qq",
+            self_id=str(event.self_id),
+            channel_id=str(event.group_id),
+            target_id=str(event.user_id),
+        )
+    if isinstance(event, OBPrivateMessageEvent):
+        return LongTaskTarget(
+            platform="qq",
+            self_id=str(event.self_id),
+            channel_id="",
+            target_id=str(event.user_id),
+        )
+    if isinstance(event, ConsoleMessageEvent):
+        return LongTaskTarget(
+            platform="console",
+            self_id=str(event.self_id),
+            channel_id=str(event.channel.id),
+            target_id=str(event.user.id),
+        )
+    if isinstance(event, DCMessageEvent):
+        self_id = ""
+        if isinstance(bot, DCBot):
+            self_id = str(bot.self_id)
+        return LongTaskTarget(
+            platform="discord",
+            self_id=self_id,
+            channel_id=str(event.channel_id),
+            target_id=str(event.user_id),
+        )
+
+
+_TaskHandler = Callable[[LongTask], Coroutine[Any, Any, Any]]
+
+
+registered_long_task_handler: dict[str, _TaskHandler] = {}
+longtask_lock = asynkio.Lock()
+
+
+def handle_long_task(callback_id: str):
+    def _decorator(func: _TaskHandler):
+        assert callback_id not in registered_long_task_handler, (
+            "有长任务的 ID 出现冲突，请换个名字！"
+        )
+        registered_long_task_handler[callback_id] = func
+        return func
+
+    return _decorator
+
+
+def _load_longtask_data() -> LongTaskModuleData:
+    try:
+        txt = LONGTASK_DATA_DIR.read_text()
+        return LongTaskModuleData.model_validate_json(txt)
+    except (FileNotFoundError, ValidationError) as e:
+        logger.info(f"取得 LongTask 数据时出现问题：{e}")
+        return LongTaskModuleData(
+            to_handle={},
+            unhandled={},
+        )
+
+
+def _save_longtask_data(data: LongTaskModuleData):
+    LONGTASK_DATA_DIR.write_text(data.model_dump_json())
+
+
+@asynccontextmanager
+async def longtask_data():
+    await longtask_lock.acquire()
+    data = _load_longtask_data()
+    yield data
+    _save_longtask_data(data)
+    longtask_lock.release()
+
+
+async def create_longtask(
+    handler: str,
+    data: dict[str, Any],
+    target: LongTaskTarget,
+    deadline: datetime.datetime,
+):
+    task = LongTask(
+        uuid=str(uuid.uuid4()),
+        data_json=json.dumps(data),
+        target=target,
+        callback=handler,
+        deadline=deadline,
+    )
+
+    await task.start()
+
+    async with longtask_data() as d:
+        d.to_handle.setdefault(handler, []).append(task)
+
+    return task
+
+
+async def init_longtask():
+    async with longtask_data() as data:
+        for v in data.to_handle.values():
+            for t in v:
+                await t.start()
+
+
+DepLongTaskTarget = Annotated[LongTaskTarget, Depends(get_long_task_target)]

konabot/common/nb/wzq_conflict.py

@@ -0,0 +1,34 @@
+from typing import cast
+from nonebot import get_bot, get_plugin_config, logger
+from nonebot.adapters import Event as BaseEvent
+from nonebot.adapters.onebot.v11.event import GroupMessageEvent
+from nonebot.adapters.onebot.v11.bot import Bot as OnebotBot
+from nonebot.rule import Rule
+from pydantic import BaseModel
+
+
+class WZQConflictConfig(BaseModel):
+    wzq_bot_qq: int = 0
+
+config = get_plugin_config(WZQConflictConfig)
+
+
+async def no_wzqbot(evt: BaseEvent):
+    if config.wzq_bot_qq <= 0:
+        return True
+    if not isinstance(evt, GroupMessageEvent):
+        return True
+    gid = evt.group_id
+    sid = evt.self_id
+    bot = cast(OnebotBot, get_bot(str(sid)))
+
+    members = await bot.get_group_member_list(group_id=gid)
+
+    members = set((m.get("user_id", -1) for m in members))
+    if config.wzq_bot_qq in members:
+        return False
+    return True
+
+
+no_wzqbot_rule = Rule(no_wzqbot)
+

konabot/common/path.py

@@ -4,9 +4,18 @@ ASSETS_PATH = Path(__file__).resolve().parent.parent.parent / "assets"
 FONTS_PATH = ASSETS_PATH / "fonts"
 
 SRC_PATH = Path(__file__).resolve().parent.parent
+DATA_PATH = SRC_PATH.parent / "data"
+LOG_PATH = DATA_PATH / "logs"
 
 DOCS_PATH = SRC_PATH / "docs"
 DOCS_PATH_MAN1 = DOCS_PATH / "user"
 DOCS_PATH_MAN3 = DOCS_PATH / "lib"
 DOCS_PATH_MAN7 = DOCS_PATH / "concepts"
 DOCS_PATH_MAN8 = DOCS_PATH / "sys"
+
+if not DATA_PATH.exists():
+    DATA_PATH.mkdir()
+
+if not LOG_PATH.exists():
+    LOG_PATH.mkdir()
+

konabot/docs/user/giftool.txt

@@ -0,0 +1,59 @@
+指令介绍
+    giftool - 对 GIF 动图进行裁剪、抽帧等处理
+
+格式
+    giftool [图片] [选项]
+
+示例
+    回复一张 GIF 并发送：
+        `giftool --ss 1.5 -t 2.0`
+        从 1.5 秒处开始，截取 2 秒长度的片段。
+
+    `giftool [图片] --ss 0:10 -to 0:15`
+        截取从 10 秒到 15 秒之间的片段（支持 MM:SS 或 HH:MM:SS 格式）。
+
+    `giftool [图片] --frames:v 10`
+        将整张 GIF 均匀抽帧，最终保留 10 帧。
+
+    `giftool [图片] --ss 2 --frames:v 5`
+        从第 2 秒开始截取，并将结果抽帧为 5 帧。
+
+参数说明
+    图片（必需）
+        - 必须是 GIF 动图。
+        - 支持直接附带图片，或回复一条含 GIF 的消息后使用指令。
+
+    --ss <时间戳>（可选）
+        - 指定开始时间（单位：秒），可使用以下格式：
+            • 纯数字（如 `1.5` 表示 1.5 秒）
+            • 分秒格式（如 `1:30` 表示 1 分 30 秒）
+            • 时分秒格式（如 `0:1:30` 表示 1 分 30 秒）
+        - 默认从开头开始（0 秒）。
+
+    -t <持续时间>（可选）
+        - 指定截取的持续时间（单位：秒），格式同 --ss。
+        - 与 --ss 配合使用：截取 [ss, ss + t] 区间。
+        - 不能与 --to 同时使用。
+
+    --to <时间戳>（可选）
+        - 指定结束时间（单位：秒），格式同 --ss。
+        - 与 --ss 配合使用：截取 [ss, to] 区间。
+        - 不能与 -t 同时使用。
+
+    --frames:v <帧数>（可选）
+        - 对截取后的片段进行均匀抽帧，保留指定数量的帧。
+        - 帧数必须为正整数（> 0）。
+        - 若原始帧数 ≤ 指定帧数，则保留全部帧。
+
+    --speed <速度>（可选）
+        - 调整 gif 图的速度。若为负数，则代表倒放
+
+使用方式
+    1. 发送指令前，请确保：
+        - 消息中附带一张 GIF 动图，或
+        - 回复一条包含 GIF 动图的消息后再发送指令。
+    2. 插件会自动：
+        - 解析 GIF 的每一帧及其持续时间（duration）
+        - 根据时间参数转换为帧索引进行裁剪
+        - 如指定抽帧，则对裁剪后的片段均匀采样
+        - 生成新的 GIF 并保持原始循环设置（loop=0）

konabot/docs/user/shadertool.txt

@@ -0,0 +1,47 @@
+指令介绍  
+    shadertool - 使用 SkSL（Skia Shader Language）代码实时渲染并生成 GIF 动画
+
+格式  
+    shadertool [选项] <SkSL 代码>
+
+示例  
+    shadertool """
+    uniform float u_time;
+    uniform float2 u_resolution;
+
+    half4 main(float2 coord) {
+        return half4(
+            1.0,
+            sin((coord.y / u_resolution.y + u_time) * 3.1415926 * 2) * 0.5 + 0.5,
+            coord.x / u_resolution.x,
+            1.0
+        );
+    }
+    """
+
+参数说明  
+    SkSL 代码（必填）  
+        - 类型：字符串（建议用英文双引号包裹）  
+        - 内容：符合 SkSL 语法的片段着色器代码，必须包含 `void main()` 函数，并为 `sk_FragColor` 赋值。  
+        - 注意：插件会自动去除代码首尾的单引号或双引号，便于命令行输入。
+
+    --width <整数>（可选）  
+        - 默认值：320  
+        - 作用：输出 GIF 的宽度（像素），必须大于 0。
+
+    --height <整数>（可选）  
+        - 默认值：180  
+        - 作用：输出 GIF 的高度（像素），必须大于 0。
+
+    --duration <浮点数>（可选）  
+        - 默认值：1.0  
+        - 作用：动画总时长（秒），必须大于 0。  
+        - 限制：`duration × fps` 必须 ≥ 1 且 ≤ 100（即至少 1 帧，最多 100 帧）。
+
+    --fps <浮点数>（可选）  
+        - 默认值：15.0  
+        - 作用：每秒帧数，控制动画流畅度，必须大于 0。  
+        - 常见值：10（低配流畅）、15（默认）、24/30（电影/视频级）。
+
+使用方式  
+    直接在群聊或私聊中发送 `shadertool` 指令，附上合法的 SkSL 代码即可。

konabot/docs/user/卵总展示.txt

@@ -0,0 +1,20 @@
+指令介绍
+    卵总展示 - 让卵总举起你的图片
+
+格式
+    <引用图片> 卵总展示 [选项]
+    卵总展示 [选项] <图片>
+
+选项
+    `--whiteness <number>` 白度
+        将原图进行指数变换，以调整它的白的程度，默认为 0.0
+
+    `--black-level <number>` 黑色等级
+        将原图减淡，数值越大越淡，范围 0.0-1.0，默认 0.2
+
+    `--opacity <number>` 不透明度
+        将你的图片叠放在图片上的不透明度，默认为 0.8
+
+    `--saturation <number>` 饱和度
+        调整原图的饱和度，应该要大于 0.0，默认为 0.85
+

konabot/docs/user/发起投票.txt

@@ -0,0 +1,11 @@
+指令介绍
+    发起投票 - 发起一个投票
+
+格式
+    发起投票 <投票标题> <选项1> <选项2> ...
+
+示例
+    `发起投票 这是一个投票 A B C` 发起标题为“这是一个投票”，选项为“A”、“B”、“C”的投票
+
+说明
+    投票各个选项之间用空格分隔，选项数量为2-15项。投票的默认有效期为24小时。

konabot/docs/user/投票.txt

@@ -0,0 +1,12 @@
+指令介绍
+    投票 - 参与已发起的投票
+
+格式
+    投票 <投票ID/标题> <选项文本>
+
+示例
+    `投票 1 A` 在ID为1的投票中，投给“A”
+    `投票 这是一个投票 B` 在标题为“这是一个投票”的投票中，投给“B”
+
+说明
+    目前不支持单人多投，每个人只能投一项。

konabot/docs/user/查看投票.txt

@@ -0,0 +1,12 @@
+指令介绍
+    查看投票 - 查看已发起的投票
+
+格式
+    查看投票 <投票ID或标题>
+
+示例
+    `查看投票 1` 查看ID为1的投票
+    `查看投票 这是一个投票` 查看标题为“这是一个投票”的投票
+
+说明
+    投票在进行时，使用此命令可以看到投票的各个选项；投票结束后，则可以看到各项的票数。

konabot/docs/user/生成二维码.txt

@@ -0,0 +1,8 @@
+指令介绍
+    生成二维码 - 将文本内容转换为二维码
+
+格式
+    生成二维码 <文本内容>
+
+示例
+    `生成二维码 嗨嗨嗨` 生成扫描结果为“嗨嗨嗨”的二维码图片

konabot/docs/user/黑白.txt

@@ -0,0 +1,5 @@
+指令介绍
+    黑白 - 将图片经过一个黑白滤镜的处理
+
+示例
+    引用一个带有图片的消息，或者消息本身携带图片，然后发送「黑白」即可

konabot/plugins/auto_accept.py

@@ -0,0 +1,25 @@
+import asyncio
+import random
+from typing import cast
+
+from loguru import logger
+from nonebot import get_bot, on_request
+from nonebot.adapters.onebot.v11.event import FriendRequestEvent
+from nonebot.adapters.onebot.v11.bot import Bot as OnebotBot
+
+from konabot.common.nb.is_admin import cfg as adminConfig
+
+add_request = on_request()
+
+@add_request.handle()
+async def _(req: FriendRequestEvent):
+    bot = cast(OnebotBot, get_bot(str(req.self_id)))
+    ok_member_ls: set[int] = set()
+    for group in adminConfig.admin_qq_group:
+        members = await bot.get_group_member_list(group_id=group)
+        ok_member_ls |= cast(set[int], set((m.get("user_id") for m in members)))
+    if req.user_id in ok_member_ls:
+        await asyncio.sleep(random.randint(5, 10))
+        await req.approve(bot)
+        logger.info(f"已经自动同意 {req.user_id} 的好友请求")
+

konabot/plugins/bilibili_fetch/__init__.py

@@ -0,0 +1,39 @@
+import re
+
+from nonebot import on_message
+from nonebot_plugin_alconna import Reference, Reply, UniMsg
+
+from nonebot.adapters import Event
+
+
+matcher_fix = on_message()
+
+pattern = (
+    r"^(?:(?:av|cv)\d+|BV[a-zA-Z0-9]{10})|"
+    r"(?:b23\.tv|bili(?:22|23|33|2233)\.cn|\.bilibili\.com|QQ小程序(?:]|]|\])哔哩哔哩).{0,500}"
+)
+
+
+@matcher_fix.handle()
+async def _(msg: UniMsg, event: Event):
+    to_search = msg.exclude(Reply, Reference).dump(json=True)
+    to_search2 = msg.exclude(Reply, Reference).extract_plain_text()
+    if not re.search(pattern, to_search) and not re.search(pattern, to_search2):
+        return
+
+    from nonebot_plugin_analysis_bilibili import handle_analysis
+
+    await handle_analysis(event)
+
+    # b_url: str
+    # b_page: str | None
+    # b_time: str | None
+    #
+    # from nonebot_plugin_analysis_bilibili.analysis_bilibili import extract as bilibili_extract
+    #
+    # b_url, b_page, b_time = bilibili_extract(to_search)
+    # if b_url is None:
+    #     return
+    #
+    # await matcher_fix.send(await UniMessage().text(b_url).export())
+

konabot/plugins/gen_qrcode/__init__.py

@@ -0,0 +1,71 @@
+import qrcode
+# from pyzbar.pyzbar import decode
+# from PIL import Image
+import requests
+from io import BytesIO
+
+from nonebot_plugin_alconna import (Alconna, Args, Field, MultiVar, UniMessage,
+                                    on_alconna)
+from nonebot_plugin_alconna.uniseg import UniMsg, At, Reply
+
+from konabot.common.nb.wzq_conflict import no_wzqbot_rule
+
+async def download_img(url):
+    resp = requests.get(url.replace("https://multimedia.nt.qq","http://multimedia.nt.qq")) # bim获取QQ的图片时避免SSLv3报错
+    img_bytes = BytesIO()
+    with open(img_bytes,"wb") as f:
+        f.write(resp.content)
+    return img_bytes
+
+def genqr(data):
+    qr = qrcode.QRCode(version=1,error_correction=qrcode.constants.ERROR_CORRECT_L,box_size=8,border=4)
+    qr.add_data(data)
+    qr.make(fit=True)
+    img = qr.make_image(fill_color="black", back_color="white")
+    img_bytes = BytesIO()
+    img.save(img_bytes, format="PNG")
+    return img_bytes
+
+"""
+async def recqr(url):
+    im_path = "assets/img/qrcode/2.jpg"
+    data = await download_img(url)
+    img = Image.open(im_path)
+    decoded_objects = decode(img)
+    data = ""
+    for obj in decoded_objects:
+        data += obj.data.decode('utf-8')
+    return data
+"""
+
+gqrc = on_alconna(Alconna(
+    "genqr",
+    Args["saying", MultiVar(str, '+'), Field(
+        missing_tips=lambda: "请输入你要转换为二维码的文字！"
+    )],
+    # UniMessage[]
+), use_cmd_start=True, use_cmd_sep=False, skip_for_unmatch=False, aliases={"生成二维码","genqrcode"}, rule=no_wzqbot_rule)
+
+@gqrc.handle()
+async def _(saying: list):
+    """
+    img = await draw_pt("\n".join(saying))
+    img_bytes = BytesIO()
+    img.save(img_bytes, format="PNG")
+
+    await pt.send(await UniMessage().image(raw=img_bytes).export())
+    
+    # print(saying)
+    # 二维码识别
+    if type(saying[0]) == 'image':
+        data = await recqr(saying[0].data['url'])
+        if data == "":
+            await gqrc.send("二维码图片解析失败！")
+        else:
+            await gqrc.send(recqr(saying[0].data['url']))
+    
+    # 二维码生成
+    else:
+    """
+        # genqr("\n".join(saying))
+    await gqrc.send(await UniMessage().image(raw=genqr("\n".join(saying))).export())

konabot/plugins/idiomgame/__init__.py

@@ -0,0 +1,521 @@
+import asyncio as asynkio
+import base64
+from pathlib import Path
+import secrets
+import json
+import datetime
+from typing import Literal, Optional
+from enum import Enum
+
+from loguru import logger
+from nonebot import on_message
+from nonebot.adapters import Event as BaseEvent
+from nonebot.adapters.console.event import MessageEvent as ConsoleMessageEvent
+from nonebot.adapters.discord.event import MessageEvent as DiscordMessageEvent
+from nonebot_plugin_alconna import (
+    Alconna,
+    Args,
+    Field,
+    Subcommand,
+    UniMessage,
+    UniMsg,
+    on_alconna,
+)
+
+from konabot.common.longtask import DepLongTaskTarget
+from konabot.common.path import ASSETS_PATH
+
+DATA_FILE_PATH = (
+    Path(__file__).parent.parent.parent.parent / "data" / "idiom_banned.json"
+)
+
+
+def load_banned_ids() -> list[str]:
+    if not DATA_FILE_PATH.exists():
+        return []
+    try:
+        return json.loads(DATA_FILE_PATH.read_text())
+    except Exception as e:
+        logger.warning(f"在解析成语接龙封禁文件时遇到问题：{e}")
+        return []
+
+
+def is_idiom_game_banned(group_id: str) -> bool:
+    banned_ids = load_banned_ids()
+    return group_id in banned_ids
+
+
+def add_banned_id(group_id: str):
+    banned_ids = load_banned_ids()
+    if group_id not in banned_ids:
+        banned_ids.append(group_id)
+        DATA_FILE_PATH.write_text(json.dumps(banned_ids, ensure_ascii=False, indent=4))
+
+
+def remove_banned_id(group_id: str):
+    banned_ids = load_banned_ids()
+    if group_id in banned_ids:
+        banned_ids.remove(group_id)
+        DATA_FILE_PATH.write_text(json.dumps(banned_ids, ensure_ascii=False, indent=4))
+
+
+class TryStartState(Enum):
+    STARTED = 0
+    ALREADY_PLAYING = 1
+    NO_REMAINING_TIMES = 2
+
+
+class TryStopState(Enum):
+    STOPPED = 0
+    NOT_PLAYING = 1
+
+
+class TryVerifyState(Enum):
+    VERIFIED = 0
+    NOT_IDIOM = 1
+    WRONG_FIRST_CHAR = 2
+    VERIFIED_BUT_NO_NEXT = 3
+    VERIFIED_GAME_END = 4
+
+
+class IdiomGame:
+    ALL_WORDS = []  # 所有四字词语
+    ALL_IDIOMS = []  # 所有成语
+    INSTANCE_LIST: dict[str, "IdiomGame"] = {}  # 群号对应的游戏实例
+    IDIOM_FIRST_CHAR = {}  # 成语首字字典
+
+    __inited = False
+
+    def __init__(self, group_id: str):
+        # 初始化一局游戏
+        self.group_id = ""
+        self.now_playing = False
+        self.score_board = {}
+        self.last_idiom = ""
+        self.last_char = ""
+        self.remain_playing_times = 3
+        self.last_play_date = ""
+        self.all_buff_score = 0
+        self.lock = asynkio.Lock()
+        self.remain_rounds = 0  # 剩余回合数
+        IdiomGame.INSTANCE_LIST[group_id] = self
+
+    def be_able_to_play(self) -> bool:
+        if self.last_play_date != datetime.date.today():
+            self.last_play_date = datetime.date.today()
+            self.remain_playing_times = 1
+        if self.remain_playing_times > 0:
+            self.remain_playing_times -= 1
+            return True
+        return False
+
+    def choose_start_idiom(self) -> str:
+        """
+        随机选择一个成语作为起始成语
+        """
+        self.last_idiom = secrets.choice(IdiomGame.ALL_IDIOMS)
+        self.last_char = self.last_idiom[-1]
+        if not self.is_nextable(self.last_char):
+            self.choose_start_idiom()
+        return self.last_idiom
+
+    @classmethod
+    def try_start_game(cls, group_id: str, force: bool = False) -> TryStartState:
+        cls.init_lexicon()
+        if not cls.INSTANCE_LIST.get(group_id):
+            cls(group_id)
+        instance = cls.INSTANCE_LIST[group_id]
+        if instance.now_playing:
+            return TryStartState.ALREADY_PLAYING
+        if not instance.be_able_to_play() and not force:
+            return TryStartState.NO_REMAINING_TIMES
+        instance.now_playing = True
+        return TryStartState.STARTED
+
+    def start_game(self, rounds: int = 100):
+        self.now_playing = True
+        self.remain_rounds = rounds
+        self.choose_start_idiom()
+
+    @classmethod
+    def try_stop_game(cls, group_id: str) -> TryStopState:
+        if not cls.INSTANCE_LIST.get(group_id):
+            return TryStopState.NOT_PLAYING
+        instance = cls.INSTANCE_LIST[group_id]
+        if not instance.now_playing:
+            return TryStopState.NOT_PLAYING
+        instance.now_playing = False
+        return TryStopState.STOPPED
+
+    def clear_score_board(self):
+        self.score_board = {}
+        self.last_char = ""
+
+    def get_score_board(self) -> dict:
+        return self.score_board
+
+    def get_all_buff_score(self) -> int:
+        return self.all_buff_score
+
+    async def skip_idiom(self, buff_score: int = -100) -> str:
+        """
+        跳过当前成语，选择下一个成语
+        """
+        await self.lock.acquire()
+        self._skip_idiom_async(buff_score)
+        self.lock.release()
+        return self.last_idiom
+
+    def _skip_idiom_async(self, buff_score: int = -100) -> str:
+        self.last_idiom = secrets.choice(IdiomGame.ALL_IDIOMS)
+        self.last_char = self.last_idiom[-1]
+        self.add_buff_score(buff_score)
+        return self.last_idiom
+
+    async def try_verify_idiom(self, idiom: str, user_id: str) -> TryVerifyState:
+        """
+        用户发送成语
+        """
+        await self.lock.acquire()
+        state = self._verify_idiom(idiom, user_id)
+        self.lock.release()
+        return state
+
+    def is_nextable(self, last_char: str) -> bool:
+        """
+        判断是否有成语可以接
+        """
+        return last_char in IdiomGame.IDIOM_FIRST_CHAR
+
+    def _verify_idiom(self, idiom: str, user_id: str) -> TryVerifyState:
+        # 新成语的首字应与上一条成语的尾字相同
+        if idiom[0] != self.last_char:
+            return TryVerifyState.WRONG_FIRST_CHAR
+        if idiom not in IdiomGame.ALL_IDIOMS and idiom not in IdiomGame.ALL_WORDS:
+            self.add_score(user_id, -0.1)
+            return TryVerifyState.NOT_IDIOM
+        self.last_idiom = idiom
+        self.last_char = idiom[-1]
+        self.add_score(user_id, 1)
+        self.remain_rounds -= 1
+        if self.remain_rounds <= 0:
+            self.now_playing = False
+            return TryVerifyState.VERIFIED_GAME_END
+        if not self.is_nextable(self.last_char):
+            # 没有成语可以接了，自动跳过
+            self._skip_idiom_async()
+            return TryVerifyState.VERIFIED_BUT_NO_NEXT
+        return TryVerifyState.VERIFIED
+
+    def get_user_score(self, user_id: str) -> int:
+        if user_id not in self.score_board:
+            return 0
+        return self.score_board[user_id]["score"]
+
+    def add_score(self, user_id: str, score: int):
+        if user_id not in self.score_board:
+            self.score_board[user_id] = {"name": user_id, "score": 0}
+        self.score_board[user_id]["score"] += score
+
+    def add_buff_score(self, score: int):
+        self.all_buff_score += score
+
+    def get_playing_state(self) -> bool:
+        return self.now_playing
+
+    def get_last_char(self) -> str:
+        return self.last_char
+
+    @classmethod
+    def init_lexicon(cls):
+        if cls.__inited:
+            return
+        cls.__inited = True
+
+        # 成语大表
+        with open(ASSETS_PATH / "lexicon" / "idiom.json", "r", encoding="utf-8") as f:
+            ALL_IDIOMS_INFOS = json.load(f)
+
+        # 词语大表
+        with open(ASSETS_PATH / "lexicon" / "ci.json", "r", encoding="utf-8") as f:
+            cls.ALL_WORDS = json.load(f)
+
+        COMMON_WORDS = []
+        # 读取 COMMON 词语大表
+        with open(ASSETS_PATH / "lexicon" / "common.txt", "r", encoding="utf-8") as f:
+            for line in f:
+                word = line.strip()
+                if len(word) == 4:
+                    COMMON_WORDS.append(word)
+
+        # 读取 THUOCL 成语库
+        with open(
+            ASSETS_PATH / "lexicon" / "THUOCL" / "data" / "THUOCL_chengyu.txt",
+            "r",
+            encoding="utf-8",
+        ) as f:
+            THUOCL_IDIOMS = [line.split(" ")[0].strip() for line in f]
+
+        # 读取 THUOCL 剩下的所有 txt 文件，只保留四字词
+        THUOCL_WORDS = []
+        import os
+
+        for filename in os.listdir(ASSETS_PATH / "lexicon" / "THUOCL" / "data"):
+            if filename.endswith(".txt") and filename != "THUOCL_chengyu.txt":
+                with open(
+                    ASSETS_PATH / "lexicon" / "THUOCL" / "data" / filename,
+                    "r",
+                    encoding="utf-8",
+                ) as f:
+                    for line in f:
+                        word = line.lstrip().split(" ")[0].strip()
+                        if len(word) == 4:
+                            THUOCL_WORDS.append(word)
+
+        # 只有成语的大表
+        cls.ALL_IDIOMS = [idiom["word"] for idiom in ALL_IDIOMS_INFOS] + THUOCL_IDIOMS
+        cls.ALL_IDIOMS = list(set(cls.ALL_IDIOMS))  # 去重
+
+        # 其他四字词语表，仅表示可以有这个词
+        cls.ALL_WORDS = (
+            [word for word in cls.ALL_WORDS if len(word) == 4]
+            + THUOCL_WORDS
+            + COMMON_WORDS
+        )
+        cls.ALL_WORDS = list(set(cls.ALL_WORDS))  # 去重
+
+        # 根据成语大表，划分出成语首字字典
+        for idiom in cls.ALL_IDIOMS + cls.ALL_WORDS:
+            if idiom[0] not in cls.IDIOM_FIRST_CHAR:
+                cls.IDIOM_FIRST_CHAR[idiom[0]] = []
+            cls.IDIOM_FIRST_CHAR[idiom[0]].append(idiom)
+
+
+evt = on_alconna(
+    Alconna(
+        "我要玩成语接龙",
+        Args["rounds?", int],
+    ),
+    use_cmd_start=True,
+    use_cmd_sep=False,
+    skip_for_unmatch=True,
+)
+
+
+@evt.handle()
+async def play_game(
+    event: BaseEvent,
+    target: DepLongTaskTarget,
+    force=False,
+    rounds: Optional[int] = 100,
+):
+    # group_id = str(event.get_session_id())
+    group_id = target.channel_id
+    if is_idiom_game_banned(group_id):
+        await evt.send(
+            await UniMessage().text("本群已被禁止使用成语接龙功能！").export()
+        )
+        return
+    rounds = rounds or 0
+    if rounds <= 0:
+        await evt.send(await UniMessage().text("干什么！你想玩负数局吗？").export())
+        return
+    state = IdiomGame.try_start_game(group_id, force)
+    if state == TryStartState.ALREADY_PLAYING:
+        await evt.send(
+            await UniMessage()
+            .text("当前已有成语接龙游戏在进行中，请稍后再试！")
+            .export()
+        )
+        return
+    if state == TryStartState.NO_REMAINING_TIMES:
+        await evt.send(await UniMessage().text("玩玩玩，就知道玩，快去睡觉！").export())
+        return
+    await evt.send(
+        await UniMessage()
+        .text(
+            "你小子，还真有意思！\n好，成语接龙游戏开始！我说一个成语，请大家接下去！"
+        )
+        .export()
+    )
+    instance = IdiomGame.INSTANCE_LIST[group_id]
+    instance.start_game(rounds)
+    # 发布成语
+    await evt.send(
+        await UniMessage()
+        .text(f"第一个成语：「{instance.last_idiom}」，请接！")
+        .export()
+    )
+
+
+evt = on_alconna(
+    Alconna(
+        "老子就是要玩成语接龙！！！",
+        Args["rounds?", int],
+    ),
+    use_cmd_start=True,
+    use_cmd_sep=False,
+    skip_for_unmatch=True,
+)
+
+
+@evt.handle()
+async def force_play_game(event: BaseEvent, rounds: Optional[int] = 100):
+    await play_game(event, force=True, rounds=rounds)
+
+
+async def end_game(event: BaseEvent, group_id: str):
+    instance = IdiomGame.INSTANCE_LIST[group_id]
+    result_text = UniMessage().text("游戏结束！\n最终得分榜：\n")
+    score_board = instance.get_score_board()
+    if len(score_board) == 0:
+        result_text += "无人得分！"
+    else:
+        # 按分数排序，名字用 at 的方式
+        sorted_score = sorted(
+            score_board.items(), key=lambda x: x[1]["score"], reverse=True
+        )
+        for i, (user_id, info) in enumerate(sorted_score):
+            result_text += (
+                f"{i + 1}. "
+                + UniMessage().at(user_id)
+                + f": {info['score'] + instance.get_all_buff_score()} 分\n"
+            )
+    await evt.send(await result_text.export())
+    instance.clear_score_board()
+
+
+evt = on_alconna(
+    Alconna("不玩了"), use_cmd_start=True, use_cmd_sep=False, skip_for_unmatch=True
+)
+
+
+@evt.handle()
+async def _(event: BaseEvent, target: DepLongTaskTarget):
+    # group_id = str(event.get_session_id())
+    group_id = target.channel_id
+    state = IdiomGame.try_stop_game(group_id)
+    if state == TryStopState.STOPPED:
+        # 发送好吧狗图片
+        # 打开好吧狗本地文件
+        with open(ASSETS_PATH / "img" / "dog" / "haoba_dog.jpg", "rb") as f:
+            img_data = f.read()
+        await evt.send(await UniMessage().image(raw=img_data).export())
+        await end_game(event, group_id)
+    else:
+        await evt.send(
+            await UniMessage().text("当前没有成语接龙游戏在进行中！").export()
+        )
+
+
+# 跳过
+evt = on_alconna(
+    Alconna("跳过成语"), use_cmd_start=True, use_cmd_sep=False, skip_for_unmatch=True
+)
+
+
+@evt.handle()
+async def _(target: DepLongTaskTarget):
+    # group_id = str(event.get_session_id())
+    group_id = target.channel_id
+    instance = IdiomGame.INSTANCE_LIST.get(group_id)
+    if not instance or not instance.get_playing_state():
+        return
+    await evt.send(await UniMessage().text("你们太菜了！全部扣100分！").export())
+    idiom = await instance.skip_idiom(-100)
+    await evt.send(
+        await UniMessage().text(f"重新开始，下一个成语是「{idiom}」").export()
+    )
+
+
+def get_user_info(event: BaseEvent):
+    if isinstance(event, DiscordMessageEvent):
+        user_id = str(event.author.id)
+        user_name = str(event.author.name)
+    else:
+        user_id = str(event.get_user_id())
+        user_name = str(event.get_user_id())
+    return user_id, user_name
+
+
+# 直接读取消息
+evt = on_message()
+
+
+@evt.handle()
+async def _(event: BaseEvent, msg: UniMsg, target: DepLongTaskTarget):
+    # group_id = str(event.get_session_id())
+    group_id = target.channel_id
+    instance = IdiomGame.INSTANCE_LIST.get(group_id)
+    if not instance or not instance.get_playing_state():
+        return
+    user_idiom = msg.extract_plain_text().strip()
+    user_id, user_name = get_user_info(event)
+    state = await instance.try_verify_idiom(user_idiom, user_id)
+    if state == TryVerifyState.WRONG_FIRST_CHAR:
+        return
+    if state == TryVerifyState.NOT_IDIOM:
+        await evt.send(
+            await UniMessage()
+            .at(user_id)
+            .text("接不上！这个不一样！你被扣了 0.1 分！")
+            .export()
+        )
+        return
+    await evt.send(
+        await UniMessage()
+        .at(user_id)
+        .text(f"接对了！你有 {instance.get_user_score(user_id)} 分！")
+        .export()
+    )
+    if state == TryVerifyState.VERIFIED_GAME_END:
+        await evt.send(await UniMessage().text("全部回合结束！").export())
+        await end_game(event, group_id)
+        return
+    if state == TryVerifyState.VERIFIED_BUT_NO_NEXT:
+        await evt.send(
+            await UniMessage()
+            .text("但是，这是条死路！你们全部都要扣 100 分！")
+            .export()
+        )
+        await evt.send(
+            await UniMessage().text(f"重新抽取成语「{instance.last_idiom}」").export()
+        )
+    await evt.send(
+        await UniMessage()
+        .text(f"下一个成语请以「{instance.get_last_char()}」开头！")
+        .export()
+    )
+
+
+evt = on_alconna(
+    Alconna("禁止成语接龙"),
+    use_cmd_start=True,
+    use_cmd_sep=False,
+    skip_for_unmatch=True,
+)
+
+
+@evt.handle()
+async def _(event: BaseEvent, target: DepLongTaskTarget):
+    # group_id = str(event.get_session_id())
+    group_id = target.channel_id
+    add_banned_id(group_id)
+    await evt.send(await UniMessage().text("本群已被禁止使用成语接龙功能！").export())
+
+
+evt = on_alconna(
+    Alconna("开启成语接龙"),
+    use_cmd_start=True,
+    use_cmd_sep=False,
+    skip_for_unmatch=True,
+)
+
+
+@evt.handle()
+async def _(event: BaseEvent, target: DepLongTaskTarget):
+    # group_id = str(event.get_session_id())
+    group_id = target.channel_id
+    remove_banned_id(group_id)
+    await evt.send(await UniMessage().text("本群已开启成语接龙功能！").export())
+

konabot/plugins/idiomgame/base/path.py

@@ -0,0 +1,3 @@
+from pathlib import Path
+
+ASSETS = Path(__file__).parent.parent / "assets"

konabot/plugins/image_process/__init__.py

@@ -1,6 +1,12 @@
+import re
+from io import BytesIO
+
 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
@@ -11,3 +17,195 @@ 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）")
+
+    frames = []
+    durations = []
+    total_duration = 0.0
+
+    try:
+        for i in range(getattr(image, "n_frames")):
+            image.seek(i)
+            frames.append(image.copy())
+            duration = image.info.get("duration", 100)  # 单位：毫秒
+            durations.append(duration)
+            total_duration += duration / 1000.0  # 转为秒
+    except EOFError:
+        pass
+
+    if not frames:
+        raise BotExceptionMessage("错误：读取 GIF 帧失败")
+
+    def time_to_frame_index(target_time: float) -> int:
+        if target_time <= 0:
+            return 0
+        cum = 0.0
+        for idx, dur in enumerate(durations):
+            cum += dur / 1000.0
+            if cum >= target_time:
+                return min(idx, len(frames) - 1)
+        return len(frames) - 1
+    start_frame = 0
+    end_frame = len(frames) - 1
+    if ss is not None:
+        start_frame = time_to_frame_index(ss)
+    if to is not None:
+        end_frame = time_to_frame_index(to)
+        if end_frame < start_frame:
+            end_frame = start_frame
+    elif t is not None:
+        end_time = (ss or 0.0) + t
+        end_frame = time_to_frame_index(end_time)
+        if end_frame < start_frame:
+            end_frame = start_frame
+
+    start_frame = max(0, start_frame)
+    end_frame = min(len(frames) - 1, end_frame)
+    selected_frames = frames[start_frame : end_frame + 1]
+    selected_durations = durations[start_frame : end_frame + 1]
+
+    if frame_count is not None and frame_count > 0:
+        if frame_count >= len(selected_frames):
+            pass
+        else:
+            step = len(selected_frames) / frame_count
+            sampled_frames = []
+            sampled_durations = []
+            for i in range(frame_count):
+                idx = int(i * step)
+                sampled_frames.append(selected_frames[idx])
+                sampled_durations.append(
+                    sum(selected_durations) // len(selected_durations)
+                )
+            selected_frames = sampled_frames
+            selected_durations = sampled_durations
+
+    output_img = BytesIO()
+
+    adjusted_durations = [
+        dur / speed_factor for dur in selected_durations
+    ]
+
+    rframes = []
+    rdur = []
+
+    acc_mod_20 = 0
+
+    for i in range(len(selected_frames)):
+        fr = selected_frames[i]
+        du: float = adjusted_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(selected_frames) > 1:
+        rframes.append(selected_frames[max(2, len(selected_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]
+
+    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())

konabot/plugins/longtask_core.py

@@ -0,0 +1,50 @@
+import asyncio
+
+# import datetime
+from loguru import logger
+import nonebot
+
+# from nonebot.adapters import Bot, Event
+# from nonebot_plugin_alconna import UniMessage
+from konabot.common.longtask import (
+    # DepLongTaskTarget,
+    # LongTask,
+    # create_longtask,
+    # get_long_task_target,
+    # handle_long_task,
+    init_longtask,
+)
+
+
+driver = nonebot.get_driver()
+INIT_FLAG = {"flag": False}
+
+
+@driver.on_bot_connect
+async def _():
+    if INIT_FLAG["flag"]:
+        return
+    INIT_FLAG["flag"] = True
+    logger.info("有 Bot 连接，等待 5 秒后初始化 LongTask 模块")
+    await asyncio.sleep(5)
+    await init_longtask()
+    logger.info("LongTask 初始化完成")
+
+
+# cmd1 = nonebot.on_command("test114")
+#
+#
+# @handle_long_task("test_callback_001")
+# async def _(lt: LongTask):
+#     await lt.target.send_message(UniMessage().text("Hello, world!"), at=True)
+#
+#
+# @cmd1.handle()
+# async def _(target: DepLongTaskTarget):
+#     await create_longtask(
+#         handler="test_callback_001",
+#         data={},
+#         target=target,
+#         deadline=datetime.datetime.now() + datetime.timedelta(seconds=2),
+#     )
+# await target.send_message(UniMessage().text("Hello, world!"), at=True)

konabot/plugins/memepack/__init__.py

@@ -2,11 +2,11 @@ from io import BytesIO
 from typing import Iterable, cast
 
 from nonebot import on_message
-from nonebot_plugin_alconna import (Alconna, Args, Field, MultiVar, Text,
+from nonebot_plugin_alconna import (Alconna, Args, Field, Image, MultiVar, Option, Text,
                                     UniMessage, UniMsg, on_alconna)
 
-from konabot.common.nb.extract_image import extract_image_from_message
-from konabot.plugins.memepack.drawing.display import draw_cao_display
+from konabot.common.nb.extract_image import PIL_Image, extract_image_from_message
+from konabot.plugins.memepack.drawing.display import draw_cao_display, draw_snaur_display
 from konabot.plugins.memepack.drawing.saying import (draw_cute_ten,
                                                      draw_geimao, draw_mnk,
                                                      draw_pt, draw_suan)
@@ -139,3 +139,24 @@ async def _(msg: UniMsg, evt: Event, bot: Bot):
                     .text(err)
                     .export()
             )
+
+
+snaur_display_cmd = on_alconna(Alconna(
+    "卵总展示",
+    Option("--whiteness",   Args["whiteness", float],   alias=["-w"]),
+    Option("--black-level", Args["black_level", float], alias=["-b"]),
+    Option("--opacity",     Args["opacity", float],     alias=["-o"]),
+    Option("--saturation",  Args["saturation", float],  alias=["-s"]),
+    Args["image", Image | None],
+))
+
+@snaur_display_cmd.handle()
+async def _(img: PIL_Image, whiteness: float = 0.0, black_level: float = 0.2,
+            opacity: float = 0.8, saturation: float = 0.85):
+    img_processed = await draw_snaur_display(
+        img, whiteness, black_level, opacity, saturation, 
+    )
+    img_data = BytesIO()
+    img_processed.save(img_data, "PNG")
+    await snaur_display_cmd.send(await UniMessage().image(raw=img_data).export())
+

konabot/plugins/memepack/drawing/display.py

@@ -4,10 +4,12 @@ from typing import Any, cast
 import cv2
 import numpy as np
 import PIL.Image
+import PIL.ImageChops
+import PIL.ImageEnhance
 
 from konabot.common.path import ASSETS_PATH
 
-cao_image = PIL.Image.open(ASSETS_PATH / "img" / "meme" / "caoimg1.png")
+cao_image       = PIL.Image.open(ASSETS_PATH / "img" / "meme" / "caoimg1.png")
 CAO_QUAD_POINTS = np.float32(cast(Any, [
     [392, 540],
     [577, 557],
@@ -15,6 +17,16 @@ CAO_QUAD_POINTS = np.float32(cast(Any, [
     [381, 687],
 ]))
 
+snaur_image_base  = PIL.Image.open(ASSETS_PATH / "img" / "meme" / "snaur_1_base.png")
+snaur_image_top   = PIL.Image.open(ASSETS_PATH / "img" / "meme" / "snaur_1_top.png")
+SNAUR_RATIO       = (1 / 2) ** .5
+SNAUR_QUAD_POINTS = np.float32(cast(Any, [
+    [0,   466 ],
+    [673, 471 ],
+    [640, 1196],
+    [106, 1280],
+]))
+
 def _draw_cao_display(image: PIL.Image.Image):
     src = np.array(image.convert("RGB"))
     h, w = src.shape[:2]
@@ -43,3 +55,87 @@ def _draw_cao_display(image: PIL.Image.Image):
 
 async def draw_cao_display(image: PIL.Image.Image):
     return await asyncio.to_thread(_draw_cao_display, image)
+
+
+def _draw_snaur_display(
+    image      : PIL.Image.Image,
+    whiteness  : float = 0.0    ,
+    black_level: float = 0.2    ,
+    opacity    : float = 0.8    ,
+    saturation : float = 0.85   ,
+):
+    src = np.array(image.convert("RGBA"))
+    _h, _w = src.shape[:2]
+
+    if _w / _h < SNAUR_RATIO:
+        _w_target = _w
+        _h_target = int(_w / SNAUR_RATIO)
+    else:
+        _w_target = int(_h * SNAUR_RATIO)
+        _h_target = _h
+
+    x_center = _w / 2
+    y_center = _h / 2
+
+    x1 = int(x_center - _w_target / 2)
+    x2 = int(x_center + _w_target / 2)
+    y1 = int(y_center - _h_target / 2)
+    y2 = int(y_center + _h_target / 2)
+
+    src = src[y1:y2, x1:x2, :]
+
+    h, w = src.shape[:2]
+    src_points = np.float32(cast(Any, [
+        [0, 0],
+        [w, 0],
+        [w, h],
+        [0, h],
+    ]))
+    dst_points = SNAUR_QUAD_POINTS
+    M = cv2.getPerspectiveTransform(cast(Any, src_points), cast(Any, dst_points))
+    output_size = snaur_image_top.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, 'RGBA')
+
+    r, g, b, a = result.split()
+    a = a.point(lambda p: int(p * opacity))
+    f2 = lambda p: int(
+        ((p / 255) ** (2 ** whiteness)) * 255 * (1 - black_level)
+            + 255 * black_level
+    )
+    r = r.point(f2)
+    g = g.point(f2)
+    b = b.point(f2)
+    result = PIL.Image.merge('RGBA', (r, g, b, a))
+
+    enhancer = PIL.ImageEnhance.Color(result)
+    result = enhancer.enhance(saturation)
+
+    result = PIL.ImageChops.multiply(result, snaur_image_base)
+
+    result = PIL.Image.alpha_composite(snaur_image_base, result)
+    result = PIL.Image.alpha_composite(result, snaur_image_top)
+    return result
+
+
+async def draw_snaur_display(
+    image      : PIL.Image.Image,
+    whiteness  : float = 0.0    ,
+    black_level: float = 0.2    ,
+    opacity    : float = 0.8    ,
+    saturation : float = 0.85   ,
+) -> PIL.Image.Image:
+    return await asyncio.to_thread(
+        _draw_snaur_display, image, whiteness, black_level,
+        opacity, saturation,
+    )
+

konabot/plugins/poll/__init__.py

@@ -0,0 +1,168 @@
+import json, time
+
+from nonebot.rule import Rule
+from nonebot_plugin_alconna import Alconna, Args, Field, MultiVar, on_alconna
+from nonebot.adapters.onebot.v11 import Event
+
+from konabot.common.nb.wzq_conflict import no_wzqbot_rule
+from konabot.common.path import ASSETS_PATH, DATA_PATH
+
+
+POLL_TEMPLATE_FILE = ASSETS_PATH / "json" / "poll.json"
+POLL_DATA_FILE = DATA_PATH / "poll.json"
+
+if not POLL_DATA_FILE.exists():
+    POLL_DATA_FILE.write_bytes(POLL_TEMPLATE_FILE.read_bytes())
+
+
+poll_list = json.loads(POLL_DATA_FILE.read_text())['poll']
+
+async def createpoll(title,qqid,options):
+    polllength = len(poll_list)
+    pollid = str(polllength)
+    poll_create = int(time.time())
+    poll_expiry = poll_create + 24*3600
+    polljson = {"title":title,"qq":qqid,"create":poll_create,"expiry":poll_expiry,"options":options,"polldata":{}}
+    poll_list[pollid] = polljson
+    writeback()
+    return pollid
+
+def getpolldata(pollid_or_title):
+    # 初始化“被指定的投票项目”
+    thepoll = {}
+    polnum = -1
+    # 判断是ID还是标题
+    if str.isdigit(pollid_or_title):
+        if pollid_or_title in poll_list:
+            thepoll = poll_list[pollid_or_title]
+            polnum = pollid_or_title
+        else:
+            return [{},-1]
+    else:
+        for i in poll_list:
+            if poll_list[i]["title"] == pollid_or_title:
+                thepoll = poll_list[i]
+                polnum = i
+                break
+        if polnum == -1:
+            return [{},-1]
+    return [thepoll,polnum]
+
+def writeback():
+    # file = open(poll_json_path,"w",encoding="utf-8")
+    # json.dump({'poll':poll_list},file,ensure_ascii=False,sort_keys=True)
+    POLL_DATA_FILE.write_text(json.dumps({
+        'poll': poll_list,
+    }, ensure_ascii=False, sort_keys=True))
+
+async def pollvote(polnum,optionnum,qqnum):
+    optiond = poll_list[polnum]["polldata"]
+    if optionnum not in optiond:
+        poll_list[polnum]["polldata"][optionnum] = []
+    poll_list[polnum]["polldata"][optionnum].append(qqnum)
+    writeback()
+    return
+
+poll = on_alconna(Alconna(
+    "poll",
+    Args["saying", MultiVar(str, '+'), Field(
+        missing_tips=lambda: "参数错误。用法：发起投票 <投票标题> <选项1> <选项2> ..."
+    )],
+), use_cmd_start=True, use_cmd_sep=False, skip_for_unmatch=False, aliases={"发起投票","createpoll"}, rule=no_wzqbot_rule)
+@poll.handle()
+async def _(saying: list, event: Event):
+    if (len(saying) < 3):
+        await poll.send("请提供至少两个投票选项！")
+    elif (len(saying) < 17):
+        title = saying[0]
+        saying.remove(title)
+        options = {}
+        for i in saying:
+            options[saying.index(i)] = i
+        qqid = event.get_user_id()
+        result = await createpoll(title,qqid,options)
+        await poll.send("已创建投票。回复 查看投票 "+str(result)+" 查看该投票。")
+    else:
+        await poll.send("投票选项太多了！请减少到15个选项以内。")
+
+viewpoll = on_alconna(Alconna(
+    "viewpoll",
+    Args["saying", MultiVar(str, '+'), Field(
+        missing_tips=lambda: "请指定投票ID或标题！。用法：查看投票 <投票ID或标题>"
+    )],
+), use_cmd_start=True, use_cmd_sep=False, skip_for_unmatch=False, aliases={"查看投票"}, rule=no_wzqbot_rule)
+@viewpoll.handle()
+async def _(saying: list):
+    # 参数，投票ID或者标题
+    # pollid_or_title = params[0]
+    polldata = getpolldata(saying[0])
+    # 被指定的投票项目
+    thepoll = polldata[0]
+    polnum = polldata[1]
+    if polnum == -1:
+        await viewpoll.send("该投票不存在！")
+    else:
+        # 检查投票是否已结束
+        pollended = 0
+        if time.time() > thepoll["expiry"]:
+            pollended = 1
+        # 回复内容
+        reply = "投票："+thepoll["title"]+" [ID: "+str(polnum)+"]"
+        # 如果投票已结束
+        if pollended:
+            for i in thepoll["options"]:
+                reply += "\n"
+                # 检查该选项是否有人投票
+                if i in thepoll["polldata"]:
+                    reply += "["+str(len(thepoll["polldata"][i]))+" 票]"
+                else:
+                    reply += "[0 票]"
+                reply += " "+thepoll["options"][i]
+            reply += "\n\n此投票已结束。"
+        else:
+            for i in thepoll["options"]:
+                reply += "\n"
+                reply += "- "+thepoll["options"][i]
+            # reply += "\n\n小提示：向bot私聊发送 /viewpoll "+str(polnum)+" 可查看已投票数哦！"
+            reply += "\n\n发送 投票 "+str(polnum)+" <选项文本> 即可参与投票！"
+        await viewpoll.send(reply)
+
+vote = on_alconna(Alconna(
+    "vote",
+    Args["saying", MultiVar(str, '+'), Field(
+        missing_tips=lambda: "参数错误。用法：投票 <投票ID/标题> <选项文本>"
+    )],
+), use_cmd_start=True, use_cmd_sep=False, skip_for_unmatch=False, aliases={"投票","参与投票"}, rule=no_wzqbot_rule)
+@vote.handle()
+async def _(saying: list, event: Event):
+    if (len(saying) < 2):
+        await vote.send("请指定投给哪一项！")
+    else:
+        polldata = getpolldata(saying[0])
+        # 被指定的投票项目
+        thepoll = polldata[0]
+        polnum = polldata[1]
+        if polnum == -1:
+            await viewpoll.finish("没有找到这个投票！")
+        # thepolldata = thepoll["polldata"]
+        # 查找对应的投票项
+        optionnum = -1
+        for i in thepoll["options"]:
+            if saying[1] == thepoll["options"][i]:
+                optionnum = i
+                break
+        if optionnum == -1:
+            reply = "此投票里面没有这一项！可用的选项有："
+            for i in thepoll["options"]:
+                reply += "\n"
+                reply += "- "+thepoll["options"][i]
+            await viewpoll.send(reply)
+        # 检查是否符合投票条件（该qq号是否已参与过投票、投票是否过期）
+        elif time.time() > thepoll["expiry"]:
+            await viewpoll.send("此投票已经结束！请发送 查看投票 "+polnum+" 查看结果。")
+        elif str(event.get_user_id()) in str(thepoll["polldata"]):
+            await viewpoll.send("你已参与过此投票！请在投票结束后发送 查看投票 "+polnum+" 查看结果。")
+        # 写入项目
+        else:
+            await pollvote(polnum,optionnum,event.get_user_id())
+            await viewpoll.send("投票成功！你投给了 "+saying[1])

konabot/plugins/roll_dice/roll_dice.py

@@ -309,7 +309,7 @@ async def generate_dice_image(number: str) -> BytesIO:
     if(len(text) > 50):
         output = BytesIO()
         push_image = Image.open(ASSETS_PATH / "img" / "dice" / "stick.png")
-        push_image.save(output,format='PNG')
+        push_image.save(output,format='GIF')
         output.seek(0)
         return output
 

konabot/plugins/simple_notify/__init__.py

@@ -1,8 +1,10 @@
-import asyncio
+import asyncio as asynkio
 import datetime
+import functools
 from pathlib import Path
 from typing import Any, Literal, cast
 
+import signal
 import nonebot
 import ptimeparse
 from loguru import logger
@@ -24,7 +26,9 @@ evt = on_message()
 
 (Path(__file__).parent.parent.parent.parent / "data").mkdir(exist_ok=True)
 DATA_FILE_PATH = Path(__file__).parent.parent.parent.parent / "data" / "notify.json"
-DATA_FILE_LOCK = asyncio.Lock()
+DATA_FILE_LOCK = asynkio.Lock()
+
+ASYNK_TASKS: set[asynkio.Task[Any]] = set()
 
 
 class Notify(BaseModel):
@@ -107,11 +111,20 @@ async def notify_now(notify: Notify):
     return True
 
 
-async def create_notify_task(notify: Notify, fail2remove: bool = True):
+def create_notify_task(notify: Notify, fail2remove: bool = True):
     async def mission():
         begin_time = datetime.datetime.now()
         if begin_time < notify.notify_time:
-            await asyncio.sleep((notify.notify_time - begin_time).total_seconds())
+            try:
+                await asynkio.sleep((notify.notify_time - begin_time).total_seconds())
+            except asynkio.CancelledError:
+                logger.debug("代办提醒被信号中止，任务退出")
+                return
+        else:
+            logger.warning(
+                f"期望在 {notify.notify_time} 在平台 {notify.platform} {notify.target_env}"
+                f" {notify.target} 的代办通知 {notify.notify_msg} 已经超时，将会直接通知！"
+            )
         res = await notify_now(notify)
         if fail2remove or res:
             await DATA_FILE_LOCK.acquire()
@@ -123,7 +136,7 @@ async def create_notify_task(notify: Notify, fail2remove: bool = True):
             DATA_FILE_LOCK.release()
         else:
             pass
-    return asyncio.create_task(mission())
+    return asynkio.create_task(mission())
 
 
 @evt.handle()
@@ -143,6 +156,7 @@ async def _(msg: UniMsg, mEvt: Event):
     # target_time = get_target_time(notify_time)
     try:
         target_time = ptimeparse.parse(notify_time)
+        logger.info(f"从 {notify_time} 解析出了时间：{target_time}")
     except Exception:
         logger.info(f"无法从 {notify_time} 中解析出时间")
         return
@@ -154,7 +168,7 @@ async def _(msg: UniMsg, mEvt: Event):
 
     await DATA_FILE_LOCK.acquire()
     cfg = load_notify_config()
-    
+
     if isinstance(mEvt, ConsoleMessageEvent):
         platform = "console"
         target = mEvt.get_user_id()
@@ -181,7 +195,7 @@ async def _(msg: UniMsg, mEvt: Event):
         notify_time=target_time,
         notify_msg=notify_text,
     )
-    await create_notify_task(notify)
+    create_notify_task(notify)
 
     cfg.notifies.append(notify)
     save_notify_config(cfg)
@@ -189,6 +203,7 @@ async def _(msg: UniMsg, mEvt: Event):
 
     await evt.send(await UniMessage().at(mEvt.get_user_id()).text(
         f" 了解啦！将会在 {notify.notify_time} 提醒你哦~").export())
+    logger.info(f"创建了一条于 {notify.notify_time} 的代办提醒")
 
 
 driver = nonebot.get_driver()
@@ -205,15 +220,41 @@ async def _():
 
     NOTIFIED_FLAG["task_added"] = True
 
-    await asyncio.sleep(10)
+    DELTA = 2
+    logger.info(f"第一次探测到 Bot 连接，等待 {DELTA} 秒后开始通知")
+    await asynkio.sleep(DELTA)
+
     await DATA_FILE_LOCK.acquire()
-    tasks = []
+
+    # tasks: set[asynkio.Task[Any]] = set()
     cfg = load_notify_config()
     if cfg.version == 1:
+        logger.info("将配置文件的版本升级为 2")
         cfg.version = 2
     else:
-        for notify in cfg.notifies:
-            tasks.append(create_notify_task(notify, fail2remove=False))
+        counter = 0
+        for notify in [*cfg.notifies]:
+            task = create_notify_task(notify, fail2remove=False)
+            ASYNK_TASKS.add(task)
+            task.add_done_callback(lambda self: ASYNK_TASKS.remove(self))
+            counter += 1
+        logger.info(f"成功创建了 {counter} 条代办事项")
+    save_notify_config(cfg)
     DATA_FILE_LOCK.release()
 
-    await asyncio.gather(*tasks)
+    loop = asynkio.get_running_loop()
+
+    # 解决 asynk task 没有被 cancel 的问题
+    async def shutdown(sig: signal.Signals):
+        logger.info(f"收到 {sig.name} 指令，正在关闭所有的东西")
+        for task in ASYNK_TASKS:
+            task.cancel()
+        await asynkio.gather(*ASYNK_TASKS, return_exceptions=True)
+        logger.info("所有的代办提醒 Task 都已经退出了")
+
+    for sig in (signal.SIGINT, signal.SIGTERM):
+        loop.add_signal_handler(sig, functools.partial(
+            asynkio.create_task, shutdown(sig)
+        ))
+
+    await asynkio.gather(*ASYNK_TASKS)

konabot/plugins/sksl/__init__.py

@@ -0,0 +1,43 @@
+from nonebot_plugin_alconna import Alconna, Args, Option, UniMessage, on_alconna
+
+from konabot.common.nb.exc import BotExceptionMessage
+from konabot.plugins.sksl.run_sksl import render_sksl_shader_to_gif
+
+
+cmd_run_sksl = on_alconna(Alconna(
+    "shadertool",
+    Option("--width", Args["width_", int]),
+    Option("--height", Args["height_", int]),
+    Option("--duration", Args["duration_", float]),
+    Option("--fps", Args["fps_", float]),
+    Args["code", str],
+))
+
+@cmd_run_sksl.handle()
+async def _(
+    code: str,
+    width_: int = 320,
+    height_: int = 180,
+    duration_: float = 1.0,
+    fps_: float = 15.0,
+):
+    if width_ <= 0 or height_ <= 0:
+        raise BotExceptionMessage("长宽应该大于 0")
+    if duration_ <= 0:
+        raise BotExceptionMessage("渲染时长应该大于 0")
+    if fps_ <= 0:
+        raise BotExceptionMessage("渲染帧率应该大于 0")
+    if fps_ * duration_ < 1:
+        raise BotExceptionMessage("时长太短或帧率太小，没有帧被渲染")
+    if fps_ * duration_ > 100:
+        raise BotExceptionMessage("太多帧啦！试着缩短一点时间吧！")
+    if width_ > 640 or height_ > 640:
+        raise BotExceptionMessage("最大支持 640x640 啦！不要太大啦！")
+
+    code = code.strip("\"").strip("'")
+
+    try:
+        res = await render_sksl_shader_to_gif(code, width_, height_, duration_, fps_)
+        await cmd_run_sksl.send(await UniMessage().image(raw=res).export())
+    except (ValueError, RuntimeError) as e:
+        await cmd_run_sksl.send(await UniMessage().text(f"渲染时遇到了问题：\n\n{e}").export())

konabot/plugins/sksl/run_sksl.py

@@ -0,0 +1,155 @@
+import asyncio
+import io
+import struct
+
+from loguru import logger
+import numpy as np
+import skia
+from PIL import Image
+
+
+def _pack_uniforms(uniforms_dict, width, height, time_val):
+    """
+    根据常见的教学用 uniform 布局打包字节数据
+    假设 SkSL 中 uniform 顺序为: float u_time; float2 u_resolution;
+    内存布局: [u_time(4B)][u_res_x(4B)][u_res_y(4B)] (总共 12 字节，无填充)
+    注意：为匹配 skia.RuntimeEffect 的紧凑布局，已移除 float 和 float2 之间的 4 字节填充。
+    """
+    # u_time (float) - 4 bytes
+    time_bytes = struct.pack('f', time_val)
+    
+    # u_resolution (vec2/float2) - 8 bytes
+    res_bytes = struct.pack('ff', float(width), float(height))
+    
+    # 移除填充字节，使用紧凑布局
+    return time_bytes + res_bytes
+
+
+def _render_sksl_shader_to_gif(
+    sksl_code: str,
+    width: int = 256,
+    height: int = 256,
+    duration: float = 2.0,
+    fps: float = 15,
+) -> io.BytesIO:
+    """
+    渲染 SkSL 着色器动画为 GIF（适配 skia-python >= 138）
+    """
+
+    logger.info(f"开始编译\n{sksl_code}")
+
+    runtime_effect = skia.RuntimeEffect.MakeForShader(sksl_code)
+    if runtime_effect is None:
+        # SkSL 编译失败时，尝试获取错误信息（如果 skia 版本支持）
+        error_message = ""
+        # 注意：skia-python 的错误信息捕获可能因版本而异
+        
+        # 尝试检查编译错误是否在日志中，但最直接的是抛出已知错误
+        raise ValueError("SkSL 编译失败，请检查语法")
+    
+    # --- 修复: 移除对不存在的 uniformSize() 的调用，直接使用硬编码的 12 字节尺寸 ---
+    # float (4 bytes) + float2 (8 bytes) = 12 bytes (基于 _pack_uniforms 函数的紧凑布局)
+    EXPECTED_UNIFORM_SIZE = 12
+
+    # 创建 CPU 后端 Surface
+    surface = skia.Surface(width, height)
+    
+    frames = []
+    total_frames = int(duration * fps)
+
+    for frame in range(total_frames):
+        time_val = frame / fps
+        
+        # 打包 uniform 数据
+        uniform_bytes = _pack_uniforms(None, width, height, time_val)
+        
+        # [检查] 确保打包后的字节数与期望值匹配
+        if len(uniform_bytes) != EXPECTED_UNIFORM_SIZE:
+             raise ValueError(
+                f"Uniform 数据大小不匹配！期望 {EXPECTED_UNIFORM_SIZE} 字节，实际 {len(uniform_bytes)} 字节。请检查 _pack_uniforms 函数。"
+            )
+
+        uniform_data = skia.Data.MakeWithCopy(uniform_bytes)
+
+        # 创建着色器
+        try:
+            # makeShader 的参数: uniform_data, children_shaders, child_count
+            shader = runtime_effect.makeShader(uniform_data, None, 0)
+            if shader is None:
+                # 如果 SkSL 语法正确但 uniform 匹配失败，makeShader 会返回 None
+                raise RuntimeError("着色器创建返回 None！请检查 SkSL 语法和 uniform 数据匹配。")
+        except Exception as e:
+            raise ValueError(f"着色器创建失败: {e}")
+
+        # 绘制
+        canvas = surface.getCanvas()
+        canvas.clear(skia.Color(0, 0, 0, 255))
+        
+        paint = skia.Paint()
+        paint.setShader(shader)
+        canvas.drawRect(skia.Rect.MakeWH(width, height), paint)
+        
+        # --- 修复 peekPixels() 错误：改用 readPixels() 将数据复制到缓冲区 ---
+        image = surface.makeImageSnapshot()
+        
+        # 1. 准备目标 ImageInfo (通常是 kBGRA_8888_ColorType)
+        target_info = skia.ImageInfo.Make(
+            image.width(), 
+            image.height(), 
+            skia.ColorType.kBGRA_8888_ColorType, # 目标格式，匹配 Skia 常见的输出格式
+            skia.AlphaType.kPremul_AlphaType
+        )
+        
+        # 2. 创建一个用于接收像素数据的 bytearray 缓冲区
+        pixel_data = bytearray(image.width() * image.height() * 4) # 4 bytes per pixel (BGRA)
+        
+        # 3. 将图像数据复制到缓冲区
+        success = image.readPixels(target_info, pixel_data, target_info.minRowBytes())
+
+        if not success:
+            raise RuntimeError("无法通过 readPixels() 获取图像像素")
+            
+        # 4. 转换 bytearray 到 NumPy 数组 (BGRA 顺序)
+        img_array = np.frombuffer(pixel_data, dtype=np.uint8).reshape((height, width, 4))
+        
+        # 5. BGRA 转换成 RGB 顺序 (交换 R 和 B 通道)
+        # [B, G, R, A] -> [R, G, B] (丢弃 A 通道)
+        rgb_array = img_array[:, :, [2, 1, 0]] 
+        
+        # 6. 创建 PIL Image
+        pil_img = Image.fromarray(rgb_array)
+        frames.append(pil_img)
+        # ------------------------------------------------------------------
+
+    # 生成 GIF
+    gif_buffer = io.BytesIO()
+    # 计算每帧的毫秒延迟
+    frame_duration_ms = int(1000 / fps)
+    frames[0].save(
+        gif_buffer,
+        format='GIF',
+        save_all=True,
+        append_images=frames[1:],
+        duration=frame_duration_ms,
+        loop=0, # 0 表示无限循环
+        optimize=True
+    )
+    gif_buffer.seek(0)
+    return gif_buffer
+
+
+async def render_sksl_shader_to_gif(
+    sksl_code: str,
+    width: int = 256,
+    height: int = 256,
+    duration: float = 2.0,
+    fps: float = 15,
+) -> io.BytesIO:
+    return await asyncio.to_thread(
+        _render_sksl_shader_to_gif,
+        sksl_code,
+        width,
+        height,
+        duration,
+        fps,
+    )

pyproject.toml

@@ -23,6 +23,9 @@ dependencies = [
     "opencv-python-headless (>=4.12.0.88,<5.0.0.0)",
     "returns (>=0.26.0,<0.27.0)",
     "ptimeparse (>=0.1.1,<0.2.0)",
+    "skia-python (>=138.0,<139.0)",
+    "nonebot-plugin-analysis-bilibili (>=2.8.1,<3.0.0)",
+    "qrcode (>=8.2,<9.0)",
 ]
 
 [build-system]
@@ -34,5 +37,10 @@ name = "pt-gitea-pypi"
 url = "https://gitea.service.jazzwhom.top/api/packages/Passthem/pypi/simple/"
 priority = "supplemental"
 
+[[tool.poetry.source]]
+name = "mirrors"
+url = "https://pypi.tuna.tsinghua.edu.cn/simple/"
+priority = "primary"
+
 [tool.poetry.dependencies]
 ptimeparse = {source = "pt-gitea-pypi"}