在 Tag 成功后也进行 ntfy 通知

Reviewed-on: #27

.dockerignore

@@ -1,4 +1,5 @@
 /.env
 /.git
+/data
 
 __pycache__

.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
@@ -26,6 +30,25 @@ steps:
     volumes:
       - name: docker-socket
         path: /var/run/docker.sock
+  - name: 在容器中测试插件加载
+    image: docker:dind
+    privileged: true
+    volumes:
+      - name: docker-socket
+        path: /var/run/docker.sock
+    commands:
+      - docker run --rm gitea.service.jazzwhom.top/mttu-developers/konabot:nightly-${DRONE_COMMIT_SHA} python scripts/test_plugin_load.py
+  - name: 发送构建结果到 ntfy
+    image: parrazam/drone-ntfy
+    when:
+      status: [success, failure]
+    settings:
+      url: https://ntfy.service.jazzwhom.top
+      topic: drone_ci
+      tags:
+        - drone-ci
+      token:
+        from_secret: NTFY_TOKEN
 
 volumes:
   - name: docker-socket
@@ -42,6 +65,10 @@ trigger:
     - tag
 
 steps:
+  - name: submodules
+    image: alpine/git
+    commands:
+    - git submodule update --init --recursive
   - name: 构建并推送 Release Docker 镜像
     image: plugins/docker:latest
     privileged: true
@@ -58,6 +85,17 @@ steps:
     volumes:
       - name: docker-socket
         path: /var/run/docker.sock
+  - name: 发送构建结果到 ntfy
+    image: parrazam/drone-ntfy
+    when:
+      status: [success, failure]
+    settings:
+      url: https://ntfy.service.jazzwhom.top
+      topic: drone_ci
+      tags:
+        - drone-ci
+      token:
+        from_secret: NTFY_TOKEN
 
 volumes:
   - name: docker-socket

.env.test

@@ -0,0 +1,4 @@
+ENVIRONMENT=test
+ENABLE_CONSOLE=false
+ENABLE_QQ=false
+ENABLE_DISCORD=false

.gitignore

@@ -1,3 +1,4 @@
 /.env
+/data
 
 __pycache__

.gitmodules

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

.vscode/launch.json

@@ -1,24 +0,0 @@
-{
-    "version": "0.2.0",
-    "configurations": [
-        {
-            "name": "运行 Bot 并调试（自动重载）",
-            "type": "debugpy",
-            "request": "launch",
-            "module": "watchfiles",
-            "args": [
-                "bot.main"
-            ],
-            "console": "integratedTerminal",
-            "justMyCode": true,
-            "env": {
-                "PYTHONPATH": "${workspaceFolder}"
-            },
-            "cwd": "${workspaceFolder}",
-            "presentation": {
-                "hidden": false,
-                "group": "bot"
-            }
-        }
-    ]
-}

.vscode/tasks.json

@@ -1,30 +0,0 @@
-{
-    "version": "2.0.0",
-    "tasks": [
-        {
-            "label": "Poetry: Export requirements.txt (Production)",
-            "type": "shell",
-            "command": "poetry export -f requirements.txt --output requirements.txt --without-hashes",
-            "group": "build",
-            "presentation": {
-                "reveal": "always",
-                "panel": "new"
-            },
-            "problemMatcher": [],
-            "detail": "导出生产环境依赖到 requirements.txt"
-        },
-        {
-            "label": "Bot: Run with Auto-reload",
-            "type": "shell",
-            "command": "poetry run watchfiles bot.main",
-            "group": "build",
-            "isBackground": true,
-            "presentation": {
-                "reveal": "always",
-                "panel": "new"
-            },
-            "problemMatcher": [],
-            "detail": "运行 bot 并启用自动重载功能"
-        }
-    ]
-}

Dockerfile

@@ -1,8 +1,59 @@
-FROM python:3.13-slim
+FROM python:3.13-slim AS base
+
+ENV VIRTUAL_ENV=/app/.venv \
+    PATH="/app/.venv/bin:$PATH" \
+    PLAYWRIGHT_BROWSERS_PATH=0
+
+# 安装所有都需要的底层依赖
+RUN apt-get update && \
+    apt-get install -y --no-install-recommends \
+    libfontconfig1 libgl1 libegl1 libglvnd0 mesa-vulkan-drivers at-spi2-common fontconfig \
+    libasound2-data libavahi-client3 libavahi-common-data libavahi-common3 libdatrie1 \
+    libfontenc1 libfribidi0 libgraphite2-3 libharfbuzz0b libice6 libpixman-1-0 \
+    libsm6 libthai-data libthai0 libunwind8 libxaw7 libxcb-render0 libxfont2 libxi6 \
+    libxkbfile1 libxmu6 libxpm4 libxrender1 libxt6t64 x11-common x11-xkb-utils \
+    xfonts-encodings xfonts-utils xkb-data xserver-common libnspr4 libatk1.0-0t64 \
+    libatk-bridge2.0-0t64 libatspi2.0-0t64 libxcomposite1 libxdamage1 libxfixes3 \
+    libxkbcommon0 libasound2t64 libnss3 fonts-noto-cjk fonts-noto-cjk-extra \
+    fonts-noto-color-emoji \
+    && rm -rf /var/lib/apt/lists/*
+
+
+
+FROM base AS builder
+
+# 安装构建依赖
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    build-essential cmake git \
+    && rm -rf /var/lib/apt/lists/*
+
+ENV POETRY_NO_INTERACTION=1 \
+    POETRY_VIRTUALENVS_IN_PROJECT=1 \
+    POETRY_VIRTUALENVS_CREATE=1 \
+    POETRY_CACHE_DIR=/tmp/poetry_cache
 
 WORKDIR /app
-COPY requirements.txt ./
-RUN pip install -r requirements.txt --no-deps
 
-COPY . .
+RUN pip install --no-cache-dir poetry
+
+COPY pyproject.toml poetry.lock ./
+RUN python -m poetry install --no-root && rm -rf $POETRY_CACHE_DIR
+
+
+
+FROM base AS runtime
+
+COPY --from=builder ${VIRTUAL_ENV} ${VIRTUAL_ENV}
+
+WORKDIR /app
+
+RUN python -m playwright install chromium
+
+COPY bot.py pyproject.toml .env.prod .env.test ./
+COPY assets ./assets
+COPY scripts ./scripts
+COPY konabot ./konabot
+
+ENV PYTHONPATH=/app
+
 CMD [ "python", "bot.py" ]

README.md

@@ -65,10 +65,10 @@ code .
 
 ### 运行
 
-你可以在 VSCode 的「运行与调试」窗口，启动 `运行 Bot 并调试（自动重载）` 任务来启动 Bot，也可以使用命令行手动启动 Bot：
+使用命令行手动启动 Bot：
 
 ```bash
-poetry run watchfiles bot.main
+poetry run watchfiles bot.main konabot
 ```
 
 如果你不希望自动重载，只是想运行 Bot，可以直接运行：

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": "我~是~谁~？"}}}

assets/webpage/ac/index.html

@@ -0,0 +1,76 @@
+<html>
+    <head>
+        <meta charset="UTF-8">
+        <meta name="viewport" content="width=device-width, initial-scale=1.0">
+        <title>空调炸炸排行榜</title>
+    </head>
+    <body>
+        <div class="box">
+            <div class="text">位居全球第 <span id="ranking" class="ranking">200</span>！</div>
+            <div class="text-2">您的群总共坏了 <span id="number" class="number">200</span> 台空调</div>
+            <img class="background" src="./assets/background.png" alt="空调炸炸排行榜">
+        </div>
+    </body>
+    <style>
+        .box {
+            position: relative;
+            width: 1024px;
+        }
+        .number {
+            font-size: 2em;
+            color: #ffdd00;
+            text-shadow: 3px 3px 6px rgba(0, 0, 0, 0.7);
+            font-weight: bold;
+            font-stretch: 50%;
+            max-width: 520px;
+            word-wrap: break-word;
+            line-height: 0.8em;
+        }
+        .background {
+            width: 1024px;
+        }
+        .text {
+            position: absolute;
+            top: 125px;
+            width: 100%;
+            font-size: 72px;
+            color: white;
+            text-shadow: 2px 2px 4px rgba(0, 0, 0, 0.7);
+            font-weight: bolder;
+            display: flex;
+            align-items: baseline;
+            justify-content: center;
+        }
+        .text-2 {
+            position: absolute;
+            top: 50px;
+            width: 100%;
+            font-size: 48px;
+            color: white;
+            text-shadow: 2px 2px 4px rgba(0, 0, 0, 0.7);
+            font-weight: bolder;
+            display: flex;
+            align-items: baseline;
+            justify-content: center;
+        }
+        .ranking {
+            font-size: 2em;
+            color: #ff0000;
+            -webkit-text-stroke: #ffffff 2px;
+            text-shadow: 3px 3px 6px rgba(0, 0, 0, 0.7);
+            font-weight: bold;
+            font-stretch: 50%;
+        }
+    </style>
+    <script>
+        // 从 URL 参数中获取 number 的值
+        const urlParams = new URLSearchParams(window.location.search);
+        const number = urlParams.get('number');
+        // 将 number 显示在页面上
+        document.getElementById('number').textContent = number;
+        // 从 URL 参数中获取 ranking 的值
+        const ranking = urlParams.get('ranking');
+        // 将 ranking 显示在页面上
+        document.getElementById('ranking').textContent = ranking;
+    </script>
+</html>

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,16 @@ 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():
+    if env.upper() == 'DEBUG' or env.upper() == 'DEV':
+        console_log_level = 'DEBUG'
+    else:
+        console_log_level = 'INFO'
+    init_logger(LOG_PATH, [
+        BotExceptionMessage,
+    ], console_log_level=console_log_level)
+
     nonebot.init()
 
     driver = nonebot.get_driver()
@@ -33,6 +46,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,80 @@
+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]],
+    console_log_level: str = "INFO",
+) -> 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=console_log_level,
+        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"控制台日志级别: {console_log_level}")

konabot/common/longtask.py

@@ -0,0 +1,302 @@
+from __future__ import annotations
+
+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"
+QQ_PRIVATE_CHAT_CHANNEL_PREFIX = "_CHANNEL_QQ_PRIVATE_"
+
+
+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"
+
+    @property
+    def is_private_chat(self):
+        return self.channel_id.startswith(QQ_PRIVATE_CHAT_CHANNEL_PREFIX)
+
+    async def send_message(self, msg: UniMessage | str, 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 isinstance(msg, str):
+            msg = UniMessage.text(msg)
+
+        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.startswith(QQ_PRIVATE_CHAT_CHANNEL_PREFIX) or not self.channel_id.strip():
+                # 私聊模式
+                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()))
+            return True
+        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,
+            )
+            return True
+        logger.warning(f"没有一个平台是期望的平台 PLATFORM={self.platform}")
+        return False
+
+
+class LongTask(BaseModel):
+    uuid: str
+    data_json: str
+    target: LongTaskTarget
+    callback: str
+    deadline: datetime.datetime
+
+    _aio_task: asynkio.Task | None = None
+
+    async def run(self):
+        now = datetime.datetime.now()
+        if self.deadline < now:
+            await self._run_task()
+            return
+        await asynkio.sleep((self.deadline - now).total_seconds())
+        async with longtask_data() as data:
+            if self.uuid not in data.to_handle[self.callback]:
+                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:
+                del datafile.to_handle[self.callback][self.uuid]
+                datafile.unhandled.setdefault(self.callback, []).append(self)
+
+            return
+        success = False
+        try:
+            await hdl(self)
+            success = True
+        except Exception as e:
+            logger.exception(e)
+        async with longtask_data() as datafile:
+            del datafile.to_handle[self.callback][self.uuid]
+            if not success:
+                datafile.unhandled.setdefault(self.callback, []).append(self)
+                logger.info(
+                    f"LongTask 执行失败 UUID={self.uuid} callback={self.callback}"
+                )
+            else:
+                logger.info(
+                    f"LongTask 工作完成 UUID={self.uuid} callback={self.callback}"
+                )
+
+    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, dict[str, 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=f"{QQ_PRIVATE_CHAT_CHANNEL_PREFIX}{event.self_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():
+    async with longtask_lock:
+        data = _load_longtask_data()
+        yield data
+        _save_longtask_data(data)
+
+
+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,
+    )
+
+    logger.info(f"创建了新的 LongTask UUID={task.uuid} CALLBACK={task.callback}")
+    await task.start()
+
+    async with longtask_data() as d:
+        d.to_handle.setdefault(handler, {})[task.uuid] = task
+
+    return task
+
+
+async def init_longtask():
+    counter = 0
+    req: set[str] = set()
+
+    async with longtask_data() as data:
+        for v in data.to_handle.values():
+            for t in v.values():
+                await t.start()
+                counter += 1
+                req.add(t.callback)
+
+        logger.info(f"LongTask 启动了任务 数量={counter} 期望的门类=[{','.join(req)}]")
+
+
+DepLongTaskTarget = Annotated[LongTaskTarget, Depends(get_long_task_target)]

konabot/common/nb/exc.py

@@ -0,0 +1,9 @@
+from nonebot_plugin_alconna import UniMessage
+
+
+class BotExceptionMessage(Exception):
+    def __init__(self, msg: UniMessage | str) -> None:
+        super().__init__()
+        if isinstance(msg, str):
+            msg = UniMessage().text(msg)
+        self.msg = msg

konabot/common/nb/extract_image.py

@@ -0,0 +1,157 @@
+from io import BytesIO
+from typing import Annotated
+
+import httpx
+import PIL.Image
+from loguru import logger
+import nonebot
+from nonebot.matcher import Matcher
+from nonebot.adapters import Bot, Event, Message
+from nonebot.adapters.discord import Bot as DiscordBot
+from nonebot.adapters.onebot.v11 import Bot as OnebotV11Bot
+from nonebot.adapters.onebot.v11 import Message as OnebotV11Message
+from nonebot.adapters.onebot.v11 import MessageEvent as OnebotV11MessageEvent
+import nonebot.params
+from nonebot_plugin_alconna import Image, RefNode, Reply, UniMessage
+from PIL import UnidentifiedImageError
+from returns.result import Failure, Result, Success
+
+
+async def download_image_bytes(url: str) -> Result[bytes, str]:
+    # if "/matcha/cache/" in url:
+    #     url = url.replace('127.0.0.1', '10.126.126.101')
+    logger.debug(f"开始从 {url} 下载图片")
+    async with httpx.AsyncClient() as c:
+        try:
+            response = await c.get(url)
+        except (httpx.ConnectError, httpx.RemoteProtocolError) as e:
+            return Failure(f"HTTPX 模块下载图片时出错：{e}")
+        except httpx.ConnectTimeout:
+            return Failure("下载图片失败了，网络超时了qwq")
+        if response.status_code != 200:
+            return Failure("无法下载图片，可能存在网络问题需要排查")
+        return Success(response.content)
+
+
+def bytes_to_pil(raw_data: bytes | BytesIO) -> Result[PIL.Image.Image, str]:
+    try:
+        if not isinstance(raw_data, BytesIO):
+            img_pil = PIL.Image.open(BytesIO(raw_data))
+        else:
+            img_pil = PIL.Image.open(raw_data)
+        img_pil.verify()
+        if not isinstance(raw_data, BytesIO):
+            img = PIL.Image.open(BytesIO(raw_data))
+        else:
+            raw_data.seek(0)
+            img = PIL.Image.open(raw_data)
+        return Success(img)
+    except UnidentifiedImageError:
+        return Failure("图像无法读取，可能是格式不支持orz")
+    except IOError:
+        return Failure("图像无法读取，可能是网络存在问题orz")
+
+
+async def unimsg_img_to_pil(image: Image) -> Result[PIL.Image.Image, str]:
+    if image.url is not None:
+        raw_result = await download_image_bytes(image.url)
+    elif image.raw is not None:
+        raw_result = Success(image.raw)
+    else:
+        return Failure("由于一些内部问题，下载图片失败了orz")
+
+    return raw_result.bind(bytes_to_pil)
+
+
+async def extract_image_from_qq_message(
+    msg: OnebotV11Message,
+    evt: OnebotV11MessageEvent,
+    bot: OnebotV11Bot,
+    allow_reply: bool = True,
+) -> Result[PIL.Image.Image, str]:
+    if allow_reply and (reply := evt.reply) is not None:
+        return await extract_image_from_qq_message(
+            reply.message,
+            evt,
+            bot,
+            False,
+        )
+    for seg in msg:
+        if seg.type == "reply" and allow_reply:
+            msgid = seg.data.get("id")
+            if msgid is None:
+                return Failure("消息可能太久远，无法读取到消息原文")
+            try:
+                msg2 = await bot.get_msg(message_id=msgid)
+            except Exception as e:
+                logger.warning(f"获取消息内容时出错：{e}")
+                return Failure("消息可能太久远，无法读取到消息原文")
+            msg2_data = msg2.get("message")
+            if msg2_data is None:
+                return Failure("消息可能太久远，无法读取到消息原文")
+            logger.debug("发现消息引用，递归一层")
+            return await extract_image_from_qq_message(
+                msg=OnebotV11Message(msg2_data),
+                evt=evt,
+                bot=bot,
+                allow_reply=False,
+            )
+        if seg.type == "image":
+            url = seg.data.get("url")
+            if url is None:
+                return Failure("无法下载图片，可能有一些网络问题")
+            data = await download_image_bytes(url)
+            return data.bind(bytes_to_pil)
+
+    return Failure("请在消息中包含图片，或者引用一个含有图片的消息")
+
+
+async def extract_image_from_message(
+    msg: Message,
+    evt: Event,
+    bot: Bot,
+    allow_reply: bool = True,
+) -> Result[PIL.Image.Image, str]:
+    if (
+        isinstance(bot, OnebotV11Bot) 
+        and isinstance(msg, OnebotV11Message) 
+        and isinstance(evt, OnebotV11MessageEvent)
+    ):
+        # 看起来 UniMessage 在这方面能力似乎不足，因此用 QQ 的
+        logger.debug('获取图片的路径 Fallback 到 QQ 模块')
+        return await extract_image_from_qq_message(msg, evt, bot, allow_reply)
+
+    for seg in UniMessage.of(msg, bot):
+        logger.info(seg)
+        if isinstance(seg, Image):
+            return await unimsg_img_to_pil(seg)
+        elif isinstance(seg, Reply) and allow_reply:
+            msg2 = seg.msg
+            logger.debug(f"深入搜索引用的消息：{msg2}")
+            if msg2 is None or isinstance(msg2, str):
+                continue
+            return await extract_image_from_message(msg2, evt, bot, False)
+        elif isinstance(seg, RefNode) and allow_reply:
+            if isinstance(bot, DiscordBot):
+                return Failure("暂时不支持在 Discord 中通过引用的方式获取图片")
+            else:
+                return Failure("暂时不支持在这里中通过引用的方式获取图片")
+    return Failure("请在消息中包含图片，或者引用一个含有图片的消息")
+
+
+async def _ext_img(
+    evt: Event,
+    bot: Bot,
+    matcher: Matcher,
+) -> PIL.Image.Image | None:
+    match await extract_image_from_message(evt.get_message(), evt, bot):
+        case Success(img):
+            return img
+        case Failure(err):
+            # raise BotExceptionMessage(err)
+            await matcher.send(await UniMessage().text(err).export())
+            return None
+    assert False
+
+
+PIL_Image = Annotated[PIL.Image.Image, nonebot.params.Depends(_ext_img)]

konabot/common/nb/is_admin.py

@@ -0,0 +1,34 @@
+from nonebot import get_plugin_config
+import nonebot
+import nonebot.adapters
+import nonebot.adapters.console
+import nonebot.adapters.discord
+import nonebot.adapters.onebot
+from pydantic import BaseModel
+
+
+class IsAdminConfig(BaseModel):
+    admin_qq_group: list[int] = []
+    admin_qq_account: list[int] = []
+    admin_discord_channel: list[int] = []
+    admin_discord_account: list[int] = []
+
+cfg = get_plugin_config(IsAdminConfig)
+
+
+def is_admin(event: nonebot.adapters.Event):
+    if isinstance(event, nonebot.adapters.onebot.v11.MessageEvent):
+        if event.user_id in cfg.admin_qq_account:
+            return True
+        if isinstance(event, nonebot.adapters.onebot.v11.GroupMessageEvent):
+            if event.group_id in cfg.admin_qq_group:
+                return True
+    if isinstance(event, nonebot.adapters.discord.event.MessageEvent):
+        if event.channel_id in cfg.admin_discord_channel:
+            return True
+        if event.user_id in cfg.admin_discord_account:
+            return True
+    if isinstance(event, nonebot.adapters.console.event.Event):
+        return True
+
+    return False

konabot/common/nb/match_keyword.py

@@ -0,0 +1,16 @@
+import re
+
+from nonebot_plugin_alconna import Text, UniMsg
+
+
+def match_keyword(*patterns: str | re.Pattern):
+    async def _matcher(msg: UniMsg):
+        text = msg.get(Text).extract_plain_text().strip()
+        for pattern in patterns:
+            if isinstance(pattern, str) and text == pattern:
+                return True
+            if isinstance(pattern, re.Pattern) and re.match(pattern, text):
+                return True
+        return False
+
+    return _matcher

konabot/common/nb/qq_broadcast.py

@@ -0,0 +1,30 @@
+from typing import Any, cast
+import nonebot
+from nonebot_plugin_alconna import UniMessage
+from nonebot.adapters.onebot.v11 import Bot as OBBot
+
+
+async def qq_broadcast(groups: list[str], msg: UniMessage[Any]):
+    bots: dict[str, OBBot] = {}
+
+    # group_id -> bot_id
+    availabilities: dict[str, str] = {}
+
+    for bot_id, bot in nonebot.get_bots().items():
+        if not isinstance(bot, OBBot):
+            continue
+        bots[bot_id] = bot
+        gl = await bot.get_group_list()
+        for g in gl:
+            gid = str(g.get("group_id", -1))
+            if gid in groups:
+                availabilities[gid] = bot_id
+
+    for group in groups:
+        if group in availabilities:
+            bot = bots[availabilities[group]]
+            await bot.send_group_msg(
+                group_id=int(group),
+                message=cast(Any, await msg.export(bot)),
+                auto_escape=False,
+            )

konabot/common/nb/reply_image.py

@@ -0,0 +1,13 @@
+from io import BytesIO
+
+import PIL
+import PIL.Image
+from nonebot.adapters import Bot
+from nonebot.matcher import Matcher
+from nonebot_plugin_alconna import UniMessage
+
+
+async def reply_image(matcher: type[Matcher], bot: Bot, img: PIL.Image.Image):
+    data = BytesIO()
+    img.save(data, "PNG")
+    await matcher.send(await UniMessage().image(raw=data).export(bot))

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

@@ -0,0 +1,21 @@
+from pathlib import Path
+
+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/common/username.py

@@ -0,0 +1,54 @@
+import re
+import nonebot
+
+from nonebot.adapters.onebot.v11 import Bot as OBBot
+
+
+class UsernameManager:
+    grouped_data: dict[int, dict[int, str]]
+    individual_data: dict[int, str]
+
+    def __init__(self) -> None:
+        self.grouped_data = {}
+        self.individual_data = {}
+
+    async def update(self):
+        for bot in nonebot.get_bots().values():
+            if isinstance(bot, OBBot):
+                for user in await bot.get_friend_list():
+                    uid = user["user_id"]
+                    nickname = user["nickname"]
+                    self.individual_data[uid] = nickname
+                for group in await bot.get_group_list():
+                    gid = group["group_id"]
+                    for member in await bot.get_group_member_list(group_id=gid):
+                        uid = member["user_id"]
+                        card = member.get("card", "")
+                        nickname = member.get("nickname", "")
+                        if card:
+                            self.grouped_data.setdefault(gid, {})[uid] = card
+                        if nickname:
+                            self.individual_data[uid] = nickname
+
+    def get(self, qqid: int, groupid: int | None = None) -> str:
+        if groupid is not None and groupid in self.grouped_data:
+            n = self.grouped_data[groupid].get(qqid)
+            if n is not None:
+                return n
+        if qqid in self.individual_data:
+            return self.individual_data[qqid]
+        return str(qqid)
+
+
+manager = UsernameManager()
+
+def get_username(qqid: int | str, group: int | str | None = None):
+    if isinstance(group, str):
+        group = None if not re.match(r"^\d+$", group) else int(group)
+    if isinstance(qqid, str):
+        if re.match(r"^\d+$", qqid):
+            qqid = int(qqid)
+        else:
+            return qqid
+    return manager.get(qqid, group)
+

konabot/common/web_render/__init__.py

@@ -0,0 +1,211 @@
+import asyncio
+import queue
+from typing import Any, Callable, Coroutine
+from loguru import logger
+from playwright.async_api import Page, Playwright, async_playwright, Browser, Page, BrowserContext
+
+
+PageFunction = Callable[[Page], Coroutine[Any, Any, Any]]
+
+
+class WebRenderer:
+    browser_pool: queue.Queue["WebRendererInstance"] = queue.Queue()
+    context_pool: dict[int, BrowserContext] = {} # 长期挂载的浏览器上下文池
+    page_pool: dict[str, Page] = {} # 长期挂载的页面池
+
+    @classmethod
+    async def get_browser_instance(cls) -> "WebRendererInstance":
+        if cls.browser_pool.empty():
+            instance = await WebRendererInstance.create()
+            cls.browser_pool.put(instance)
+        instance = cls.browser_pool.get()
+        cls.browser_pool.put(instance)
+        return instance
+    
+    @classmethod
+    async def get_browser_context(cls) -> BrowserContext:
+        instance = await cls.get_browser_instance()
+        if id(instance) not in cls.context_pool:
+            context = await instance.browser.new_context()
+            cls.context_pool[id(instance)] = context
+            logger.debug(f"Created new persistent browser context for WebRendererInstance {id(instance)}")
+        return cls.context_pool[id(instance)]
+
+    @classmethod
+    async def render(
+        cls,
+        url: str,
+        target: str,
+        params: dict = {},
+        other_function: PageFunction | None = None,
+        timeout: int = 30,
+    ) -> bytes:
+        '''
+        访问指定URL并返回截图
+
+        :param url: 目标URL
+        :param target: 渲染目标，如 ".box"、"#main" 等CSS选择器
+        :param timeout: 页面加载超时时间，单位秒
+        :param params: URL键值对参数
+        :param other_function: 其他自定义操作函数，接受page参数
+        :return: 截图的字节数据
+
+        '''
+        instance = await cls.get_browser_instance()
+        logger.debug(f"Using WebRendererInstance {id(instance)} to render {url} targeting {target}")
+        return await instance.render(url, target, params=params, other_function=other_function, timeout=timeout)
+
+
+    @classmethod
+    async def render_persistent_page(cls, page_id: str, url: str, target: str, params: dict = {}, other_function: callable = None, timeout: int = 30) -> bytes:
+        '''
+        使用长期挂载的页面访问指定URL并返回截图
+
+        :param page_id: 页面唯一标识符
+        :param url: 目标URL
+        :param target: 渲染目标，如 ".box"、"#main" 等CSS选择器
+        :param timeout: 页面加载超时时间，单位秒
+        :param params: URL键值对参数
+        :param other_function: 其他自定义操作函数，接受page参数
+        :return: 截图的字节数据
+
+        '''
+        logger.debug(f"Requesting persistent render for page_id {page_id} at {url} targeting {target} with timeout {timeout}")
+        instance = await cls.get_browser_instance()
+        if page_id not in cls.page_pool:
+            context = await cls.get_browser_context()
+            page = await context.new_page()
+            cls.page_pool[page_id] = page
+            logger.debug(f"Created new persistent page for page_id {page_id} using WebRendererInstance {id(instance)}")
+        page = cls.page_pool[page_id]
+        return await instance.render_with_page(page, url, target, params=params, other_function=other_function, timeout=timeout)
+    
+    @classmethod
+    async def render_file(
+        cls,
+        file_path: str,
+        target: str,
+        params: dict = {},
+        other_function: PageFunction | None = None,
+        timeout: int = 30,
+    ) -> bytes:
+        '''
+        访问指定本地文件URL并返回截图
+
+        :param file_path: 目标文件路径
+        :param target: 渲染目标，如 ".box"、"#main" 等CSS选择器
+        :param timeout: 页面加载超时时间，单位秒
+        :param params: URL键值对参数
+        :param other_function: 其他自定义操作函数，接受page参数
+        :return: 截图的字节数据
+
+        '''
+        instance = await cls.get_browser_instance()
+        logger.debug(f"Using WebRendererInstance {id(instance)} to render file {file_path} targeting {target}")
+        return await instance.render_file(file_path, target, params=params, other_function=other_function, timeout=timeout)
+
+    @classmethod
+    async def close_persistent_page(cls, page_id: str) -> None:
+        '''
+        关闭并移除长期挂载的页面
+
+        :param page_id: 页面唯一标识符
+        '''
+        if page_id in cls.page_pool:
+            page = cls.page_pool[page_id]
+            await page.close()
+            del cls.page_pool[page_id]
+            logger.debug(f"Closed and removed persistent page for page_id {page_id}")
+
+
+
+class WebRendererInstance:
+    def __init__(self):
+        self._playwright: Playwright | None = None
+        self._browser: Browser | None = None
+        self.lock = asyncio.Lock()
+
+    @property
+    def playwright(self) -> Playwright:
+        assert self._playwright is not None
+        return self._playwright
+
+    @property
+    def browser(self) -> Browser:
+        assert self._browser is not None
+        return self._browser
+
+    async def init(self):
+        self._playwright = await async_playwright().start()
+        self._browser = await self.playwright.chromium.launch(headless=True)
+
+    @classmethod
+    async def create(cls) -> "WebRendererInstance":
+        instance = cls()
+        await instance.init()
+        return instance
+
+    async def render(
+        self,
+        url: str,
+        target: str,
+        index: int = 0,
+        params: dict = {},
+        other_function: PageFunction | None = None,
+        timeout: int = 30
+    ) -> bytes:
+        '''
+        访问指定URL并返回截图
+
+        :param url: 目标URL
+        :param target: 渲染目标，如 ".box"、"#main" 等CSS选择器
+        :param timeout: 页面加载超时时间，单位秒
+        :param index: 如果目标是一个列表，指定要截图的元素索引
+        :param params: URL键值对参数
+        :param other_function: 其他自定义操作函数，接受page参数
+        :return: 截图的字节数据
+
+        '''
+        async with self.lock:
+            context = await self.browser.new_context()
+            page = await context.new_page()
+            screenshot = await self.inner_render(page, url, target, index, params, other_function, timeout)
+            await page.close()
+            await context.close()
+            return screenshot
+
+    async def render_with_page(self, page: Page, url: str, target: str, index: int = 0, params: dict = {}, other_function: callable = None, timeout: int = 30) -> bytes:
+        async with self.lock:
+            screenshot = await self.inner_render(page, url, target, index, params, other_function, timeout)
+            return screenshot
+        
+    async def render_file(self, file_path: str, target: str, index: int = 0, params: dict = {}, other_function: callable = None, timeout: int = 30) -> bytes:
+        file_path = "file:///" + str(file_path).replace("\\", "/")
+        return await self.render(file_path, target, index, params, other_function, timeout)
+
+    async def inner_render(self, page: Page, url: str, target: str, index: int = 0, params: dict = {}, other_function: callable = None, timeout: int = 30) -> bytes:
+        logger.debug(f"Navigating to {url} with timeout {timeout}")
+        url_with_params = url + ("?" + "&".join(f"{k}={v}" for k, v in params.items()) if params else "")
+        await page.goto(url_with_params, timeout=timeout * 1000, wait_until="load")
+        logger.debug("Page loaded successfully")
+        # 等待目标元素出现
+        await page.wait_for_selector(target, timeout=timeout * 1000)
+        logger.debug(f"Target element '{target}' found, taking screenshot")
+        if other_function:
+            await other_function(page)
+        elements = await page.query_selector_all(target)
+        if not elements:
+            logger.error(f"Target element '{target}' not found on the page.")
+            return None
+        if index >= len(elements):
+            logger.error(f"Index {index} out of range for elements matching '{target}'")
+            return None
+        element = elements[index]
+        screenshot = await element.screenshot()
+        logger.debug(f"Screenshot taken successfully")
+        return screenshot
+
+    async def close(self):
+        await self.browser.close()
+        await self.playwright.stop()
+

konabot/docs/README.md

@@ -0,0 +1,40 @@
+# 此方 Bot 的文档系统
+
+此方 Bot 使用类 Linux 的 `man` 指令来管理文档。文档一般建议使用纯文本书写，带有相对良好的格式。
+
+## 文件夹摆放规则
+
+`docs` 目录下，有若干文档可以拿来阅读和输出。每个子文件夹里，文档文件使用名字不含空格的 txt 文件书写，其他后缀名的文件将会被忽略。所以，如果你希望有些文件只在代码库中可阅读，你可以使用 `.md` 格式。
+
+### 1 - user
+
+`docs/user` 目录下的文档是直接会给用户进行检索的文档，在直接使用 `man` 指令时，会搜索该文件夹的全部文件，以知晓所有有文档的指令。
+
+### 3 - lib
+
+`docs/lib` 目录下的文档主要给该项目的维护者进行阅读和使用，讲述的是本项目内置的一些函数的功能讲解（一般以便利为主要目的）以及一些项目安排上的要求。一般不会列举，除非用户指定要求列举该范围。
+
+### 7 - concepts
+
+`docs/concepts` 用来摆放任何的概念。任何的。一般不会列举，除非用户指定要求列举该范围。
+
+### 8 - sys
+
+`docs/sys` 用于摆放仅 MTTU 群可以使用的文档集合。在 MTTU 群内，该目录下的文档也会被索引，否则文档将不可阅读。
+
+## 书写规范
+
+无特殊要求，因为当用户进行 `man` 的时候，会将文档内的内容原封不动地展示出来。但是，你仍然可以模仿 Linux 下的 `man` 指令的格式进行书写。
+
+```
+指令介绍
+    man - 用于展示此方 BOT 使用手册的指令
+
+格式
+    man [文档类型] <指令>
+
+示例
+    `man` 查看所有有文档的指令清单
+    `man 喵` 查看指令「喵」的使用说明
+……
+```

konabot/docs/concepts/罗文.txt

@@ -0,0 +1,2 @@
+关于罗文和洛温：
+    AdoreLowen 希望和洛温阿特金森区分，所以最好就不要叫他洛温了！此方 BOT 会在一些群提醒叫错了的人。

konabot/docs/lib/is_admin.txt

@@ -0,0 +1,45 @@
+指令介绍
+    is_admin - 用于判断当前事件是否来自管理员的内部权限校验函数
+
+格式
+    from konabot.common.nb.is_admin import is_admin
+    from nonebot import on
+    from nonebot.adapters import Event
+    from loguru import logger
+
+    @on().handle()
+    async def _(event: Event):
+        if is_admin(event):
+            logger.info("管理员发送了消息")
+
+说明
+    is_admin 是 Bot 内部用于权限控制的核心函数，根据事件来源（QQ、Discord、控制台）及插件配置，判断触发事件的用户或群组是否具有管理员权限。
+
+支持的适配器与判定逻辑：
+    • OneBot V11（QQ）
+        - 若用户 ID 在配置项 admin_qq_account 中，则视为管理员
+        - 若为群聊消息，且群 ID 在配置项 admin_qq_group 中，则视为管理员
+
+    • Discord
+        - 若频道 ID 在配置项 admin_discord_channel 中，则视为管理员
+        - 若用户 ID 在配置项 admin_discord_account 中，则视为管理员
+
+    • Console（控制台）
+        - 所有控制台输入均默认视为管理员操作，自动返回 True
+
+配置项（位于插件配置中）
+    ADMIN_QQ_GROUP: list[int]
+        允许的管理员 QQ 群 ID 列表
+
+    ADMIN_QQ_ACCOUNT: list[int]
+        允许的管理员 QQ 账号 ID 列表
+
+    ADMIN_DISCORD_CHANNEL: list[int]
+        允许的管理员 Discord 频道 ID 列表
+
+    ADMIN_DISCORD_ACCOUNT: list[int]
+        允许的管理员 Discord 用户 ID 列表
+
+注意事项
+    - 若未在配置文件中设置任何管理员 ID，该函数对所有非控制台事件返回 False
+    - 控制台事件始终拥有管理员权限，便于本地调试与运维

konabot/docs/sys/konaph.txt

@@ -0,0 +1,4 @@
+指令介绍
+    konaph - KonaBot 的 PuzzleHunt 管理工具
+
+详细介绍请直接输入 konaph 获取使用指引（该指令权限仅对部分人开放。如果你有权限的话才有响应。建议在此方 BOT 私聊使用该指令。）

konabot/docs/sys/out.txt

@@ -0,0 +1 @@
+MAN what can I say!

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/man.txt

@@ -0,0 +1,20 @@
+指令介绍
+    man - 用于展示此方 BOT 使用手册的指令
+
+格式
+    man 文档类型
+    man [文档类型] <指令>
+
+示例
+    `man` 查看所有有文档的指令清单
+    `man 3` 列举所有可读文档的库函数清单
+    `man 喵` 查看指令「喵」的使用说明
+    `man 8 out` 查看管理员指令「out」的使用说明
+
+文档类型
+    文档类型用来区分同一指令在不同场景下的情景。你可以使用数字编号进行筛选。分为这些种类：
+
+    - 1 用户态指令，用于日常使用的指令
+    - 3 库函数指令，用于 Bot 开发用的函数查询
+    - 7 概念指令，用于概念解释
+    - 8 系统指令，仅管理员可用

konabot/docs/user/ntfy.txt

@@ -0,0 +1,15 @@
+指令介绍
+    ntfy - 配置使用 ntfy 来更好地为你通知此方 BOT 代办
+
+指令示例
+    `ntfy 创建`
+        创建一个随机的 ntfy 订阅主题来提醒代办，此方 Bot 将会给你使用指引。你可以前往 https://ntfy.sh/ 官网下载 ntfy APP，或者使用网页版 ntfy。
+
+    `ntfy 创建 kagami-notice`
+        创建一个名字含有 kagami-notice 的 ntfy 订阅主题
+
+    `ntfy 删除`
+        清除并不再使用 ntfy 向你通知
+
+另见
+    提醒我(1) 查询提醒(1) 删除提醒(1)

konabot/docs/user/openssl.txt

@@ -0,0 +1,21 @@
+指令介绍
+    openssl - 用于生成指定长度的加密安全随机数据
+
+格式
+    openssl rand <模式> <字节数>
+
+示例
+    `openssl rand -hex 16` 生成 16 字节的十六进制随机数
+    `openssl rand -base64 32` 生成 32 字节并以 Base64 编码输出的随机数据
+
+说明
+    该指令使用 Python 的 secrets 模块生成加密安全的随机字节，并支持以十六进制（-hex）或 Base64（-base64）格式输出。
+
+参数说明
+    模式（mode）
+        - -hex     ：以十六进制字符串形式输出随机数据
+        - -base64  ：以 Base64 编码字符串形式输出随机数据
+
+    字节数（num）
+        - 必须为正整数
+        - 最大支持 256 字节

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/ytpgif.txt

@@ -0,0 +1,41 @@
+指令介绍
+    ytpgif - 生成来回镜像翻转的仿 YTPMV 动图
+
+格式
+    ytpgif [倍速]
+
+示例
+    `ytpgif`  
+        使用默认倍速（1.0）处理你发送或回复的图片，生成镜像动图。
+
+    `ytpgif 2.5`  
+        以 2.5 倍速处理图片，生成更快节奏的镜像动图。
+
+    回复一张图片并发送 `ytpgif 0.5`  
+        以慢速（0.5 倍）生成镜像动图。
+
+参数说明
+    倍速（可选）
+        - 类型：浮点数
+        - 默认值：1.0
+        - 有效范围：0.1 ～ 20.0
+        - 作用：
+            • 对于静态图：控制镜像切换的快慢（值越大，切换越快）。
+            • 对于动图：控制截取原始动图正向和反向片段的时长（值越大，截取的片段越长）。
+
+使用方式
+    发送指令前，请确保：
+      - 直接在消息中附带一张图片，或
+      - 回复一条包含图片的消息后再发送指令。
+
+    插件会自动：
+      - 下载并识别图片（支持静态图和 GIF 动图）
+      - 自动缩放至最大边长不超过 256 像素（保持宽高比）
+      - 静态图 → 生成“原图↔镜像”循环动图
+      - 动图 → 截取开头一段正向播放 + 同一段镜像翻转播放，拼接成新动图
+      - 保留透明通道（如原图含透明），否则转为 RGB 避免颜色异常
+
+注意事项
+    - 图片过大、格式损坏或网络问题可能导致处理失败。
+    - 动图帧数过多或单帧过短可能无法生成有效输出。
+    - 输出 GIF 最大单段帧数限制为 500 帧，以防资源耗尽。

konabot/docs/user/删除提醒.txt

@@ -0,0 +1,8 @@
+指令介绍
+    删除提醒 - 删除在`查询提醒(1)`中查到的提醒
+
+指令示例
+    `删除提醒 1` 在查询提醒后，删除编号为 1 的提醒
+
+另见
+    提醒我(1) 查询提醒(1) ntfy(1)

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,2 @@
+指令介绍
+    喵 - 你发喵，此方就会回复喵

konabot/docs/user/投票.txt

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

konabot/docs/user/提醒我.txt

@@ -0,0 +1,15 @@
+指令介绍
+    提醒我 - 在指定的时间提醒人事项的工具
+
+使用示例
+    `下午五点提醒我吃饭`
+        创建一个下午五点的提醒，提醒你吃饭
+
+    `两分钟后提醒我睡觉`
+        创建一个相对于现在推迟 2 分钟的提醒，提醒你睡觉
+
+    `2026年4月25日20点整提醒我生日快乐`
+        创建一个指定日期和时间的提醒
+
+另见
+    查询提醒(1) 删除提醒(1) ntfy(1)

konabot/docs/user/摇数字.txt

@@ -0,0 +1,7 @@
+指令介绍
+    摇数字 - 生成一个随机数字并发送
+
+示例
+    `摇数字` 随机生成一个 1-6 的数字
+
+该指令不接受任何参数，直接调用即可。

konabot/docs/user/摇骰子.txt

@@ -0,0 +1,22 @@
+指令介绍
+    摇骰子 - 用于生成随机数并以骰子图像形式展示的指令
+
+格式
+    摇骰子 [最小值] [最大值]
+
+示例
+    `摇骰子` 随机生成一个 1-6 的数字，并显示对应的骰子图像
+    `摇骰子 10` 生成 1 到 10 之间的随机整数
+    `摇骰子 0.5` 生成 0 到 0.5 之间的随机小数
+    `摇骰子 -5 5` 生成 -5 到 5 之间的随机数
+
+说明
+    该指令支持以下几种调用方式：
+      - 不带参数：使用默认范围生成随机数
+      - 仅指定一个参数 f1：
+          - 若 f1 > 1，则生成 [1, f1] 范围内的随机数
+          - 若 0 < f1 ≤ 1，则生成 [0, f1] 范围内的随机数
+          - 若 f1 ≤ 0，则生成 [f1, 0] 范围内的随机数
+      - 指定两个参数 f1 和 f2：生成 [f1, f2] 范围内的随机数（顺序无关，内部会自动处理大小）
+
+返回结果将以骰子样式的图像形式展示生成的随机数值。

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

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

konabot/docs/user/查询提醒.txt

@@ -0,0 +1,9 @@
+指令介绍
+    查询提醒 - 查询已经创建的提醒
+
+指令格式
+    `查询提醒` 查询提醒
+    `查询提醒 2` 查询第二页提醒
+
+另见
+    提醒我(1) 删除提醒(1) ntfy(1)

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

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

konabot/docs/user/雷达回波.txt

@@ -0,0 +1,13 @@
+指令介绍
+    雷达回波 - 用于获取指定地区的天气雷达回波图像
+
+格式
+    雷达回波 <地区>
+
+示例
+    `雷达回波 华南` 获取华南地区的天气雷达回波图
+    `雷达回波 全国` 获取全国的天气雷达回波图
+
+说明
+    该指令通过查询中国气象局 https://www.nmc.cn/publish/radar/chinaall.html ，获取指定地区的实时天气雷达回波图像。
+    支持的地区有：全国 华北 东北 华东 华中 华南 西南 西北。

konabot/docs/user/黑白.txt

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

konabot/plugins/air_conditioner/__init__.py

@@ -0,0 +1,162 @@
+from io import BytesIO
+from typing import Optional, Union
+import cv2
+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, AlconnaMatcher, Args, UniMessage, on_alconna
+from PIL import Image
+import numpy as np
+from konabot.common.longtask import DepLongTaskTarget
+from konabot.common.path import ASSETS_PATH
+from konabot.common.web_render import WebRenderer
+from konabot.plugins.air_conditioner.ac import AirConditioner, CrashType, generate_ac_image, wiggle_transform
+
+import random
+import math
+
+def get_ac(id: str) -> AirConditioner:
+    ac = AirConditioner.air_conditioners.get(id)
+    if ac is None:
+        ac = AirConditioner(id)
+    return ac
+
+async def send_ac_image(event: type[AlconnaMatcher], ac: AirConditioner):
+    if(ac.burnt == True):
+        # 打开坏掉的空调图片
+        with open(ASSETS_PATH / "img" / "ac" / "broken_ac.png", "rb") as f:
+            # 将其转为 GIF 格式发送
+            output = BytesIO()
+            Image.open(f).save(output, format="GIF")
+            output.seek(0)
+            await event.send(await UniMessage().image(raw=output).export())
+        return
+    if(ac.frozen == True):
+        # 打开坏掉的空调图片
+        with open(ASSETS_PATH / "img" / "ac" / "frozen_ac.png", "rb") as f:
+            # 将其转为 GIF 格式发送
+            output = BytesIO()
+            Image.open(f).save(output, format="GIF")
+            output.seek(0)
+            await event.send(await UniMessage().image(raw=output).export())
+        return
+    ac_image = await generate_ac_image(ac)
+    await event.send(await UniMessage().image(raw=ac_image).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):
+    id = target.channel_id
+    ac = get_ac(id)
+    await send_ac_image(evt, ac)
+
+evt = on_alconna(Alconna(
+    "开空调"
+), use_cmd_start=True, use_cmd_sep=False, skip_for_unmatch=True)
+
+@evt.handle()
+async def _(event: BaseEvent, target: DepLongTaskTarget):
+    id = target.channel_id
+    ac = get_ac(id)
+    ac.on = True
+    await send_ac_image(evt, ac)
+
+evt = on_alconna(Alconna(
+    "关空调"
+), use_cmd_start=True, use_cmd_sep=False, skip_for_unmatch=True)
+
+@evt.handle()
+async def _(event: BaseEvent, target: DepLongTaskTarget):
+    id = target.channel_id
+    ac = get_ac(id)
+    ac.on = False
+    await send_ac_image(evt, ac)
+
+evt = on_alconna(Alconna(
+    "空调升温",
+    Args["temp?", Optional[Union[int, float]]]  # 可选参数，升温的度数，默认为1
+), use_cmd_start=True, use_cmd_sep=False, skip_for_unmatch=True)
+
+@evt.handle()
+async def _(event: BaseEvent, target: DepLongTaskTarget, temp: Optional[Union[int, float]] = 1):
+    if temp <= 0:
+        return
+    id = target.channel_id
+    ac = get_ac(id)
+    if not ac.on or ac.burnt == True or ac.frozen == True:
+        await send_ac_image(evt, ac)
+        return
+    ac.temperature += temp
+    if ac.temperature > 40:
+        # 根据温度随机出是否爆炸，40度开始，呈指数增长
+        possibility = -math.e ** ((40-ac.temperature) / 50) + 1
+        if random.random() < possibility:
+            # 打开爆炸图片
+            with open(ASSETS_PATH / "img" / "other" / "boom.jpg", "rb") as f:
+                output = BytesIO()
+                # 爆炸抖动
+                frames = wiggle_transform(np.array(Image.open(f)), intensity=5)
+                pil_frames = [Image.fromarray(frame) for frame in frames]
+                pil_frames[0].save(output, format="GIF", save_all=True, append_images=pil_frames[1:], loop=0, duration=35, disposal=2)
+                output.seek(0)
+                await evt.send(await UniMessage().image(raw=output).export())
+            ac.broke_ac(CrashType.BURNT)
+            await evt.send("太热啦，空调炸了！")
+            return
+    await send_ac_image(evt, ac)
+
+evt = on_alconna(Alconna(
+    "空调降温",
+    Args["temp?", Optional[Union[int, float]]]  # 可选参数，降温的度数，默认为1
+), use_cmd_start=True, use_cmd_sep=False, skip_for_unmatch=True)
+
+@evt.handle()
+async def _(event: BaseEvent, target: DepLongTaskTarget, temp: Optional[Union[int, float]] = 1):
+    if temp <= 0:
+        return
+    id = target.channel_id
+    ac = get_ac(id)
+    if not ac.on or ac.burnt == True or ac.frozen == True:
+        await send_ac_image(evt, ac)
+        return
+    ac.temperature -= temp
+    if ac.temperature < 0:
+        # 根据温度随机出是否冻结，0度开始，呈指数增长
+        possibility = -math.e ** (ac.temperature / 50) + 1
+        if random.random() < possibility:
+            ac.broke_ac(CrashType.FROZEN)
+    await send_ac_image(evt, ac)
+
+evt = on_alconna(Alconna(
+    "换空调"
+), use_cmd_start=True, use_cmd_sep=False, skip_for_unmatch=True)
+
+@evt.handle()
+async def _(event: BaseEvent, target: DepLongTaskTarget):
+    id = target.channel_id
+    ac = get_ac(id)
+    ac.change_ac()
+    await send_ac_image(evt, ac)
+
+evt = on_alconna(Alconna(
+    "空调炸炸排行榜",
+), use_cmd_start=True, use_cmd_sep=False, skip_for_unmatch=True)
+
+@evt.handle()
+async def _(event: BaseEvent, target: DepLongTaskTarget):
+    id = target.channel_id
+    ac = get_ac(id)
+    number, ranking = ac.get_crashes_and_ranking()
+    params = {
+        "number": number,
+        "ranking": ranking
+    }
+    image = await WebRenderer.render_file(
+        file_path=ASSETS_PATH / "webpage" / "ac" / "index.html",
+        target=".box",
+        params=params
+    )
+    await evt.send(await UniMessage().image(raw=image).export())

konabot/plugins/air_conditioner/ac.py

@@ -0,0 +1,288 @@
+from enum import Enum
+from io import BytesIO
+
+import cv2
+import numpy as np
+from PIL import Image, ImageDraw, ImageFont
+
+from konabot.common.path import ASSETS_PATH, FONTS_PATH
+from konabot.common.path import DATA_PATH
+import json
+
+class CrashType(Enum):
+    BURNT = 0
+    FROZEN = 1
+
+class AirConditioner:
+    air_conditioners: dict[str, "AirConditioner"] = {}
+
+    def __init__(self, id: str) -> None:
+        self.id = id
+        self.on = False
+        self.temperature = 24  # 默认温度
+        self.burnt = False
+        self.frozen = False
+        AirConditioner.air_conditioners[id] = self
+
+    def change_ac(self):
+        self.burnt = False
+        self.frozen = False
+        self.on = False
+        self.temperature = 24  # 重置为默认温度
+
+    def broke_ac(self, crash_type: CrashType):
+        '''
+        让空调坏掉，并保存数据
+
+        :param crash_type: CrashType 枚举，表示空调坏掉的类型
+        '''
+        match crash_type:
+            case CrashType.BURNT:
+                self.burnt = True
+            case CrashType.FROZEN:
+                self.frozen = True
+        self.save_crash_data(crash_type)
+
+    def save_crash_data(self, crash_type: CrashType):
+        '''
+        如果空调爆炸了，就往本地的 ac_crash_data.json 里该 id 的记录加一
+        '''
+        data_file = DATA_PATH / "ac_crash_data.json"
+        crash_data = {}
+        if data_file.exists():
+            with open(data_file, "r", encoding="utf-8") as f:
+                crash_data = json.load(f)
+        if self.id not in crash_data:
+            crash_data[self.id] = {"burnt": 0, "frozen": 0}
+        match crash_type:
+            case CrashType.BURNT:
+                crash_data[self.id]["burnt"] += 1
+            case CrashType.FROZEN:
+                crash_data[self.id]["frozen"] += 1
+        with open(data_file, "w", encoding="utf-8") as f:
+            json.dump(crash_data, f, ensure_ascii=False, indent=4)
+
+    def get_crashes_and_ranking(self) -> tuple[int, int]:
+        '''
+        获取该群在全国空调损坏的数量与排行榜的位置
+        '''
+        data_file = DATA_PATH / "ac_crash_data.json"
+        if not data_file.exists():
+            return 0, 1
+        with open(data_file, "r", encoding="utf-8") as f:
+            crash_data = json.load(f)
+        ranking_list = []
+        for gid, record in crash_data.items():
+            total = record.get("burnt", 0) + record.get("frozen", 0)
+            ranking_list.append((gid, total))
+        ranking_list.sort(key=lambda x: x[1], reverse=True)
+        total_crashes = crash_data.get(self.id, {}).get("burnt", 0) + crash_data.get(self.id, {}).get("frozen", 0)
+        rank = 1
+        for gid, total in ranking_list:
+            if gid == self.id:
+                break
+            rank += 1
+        return total_crashes, rank
+
+def text_to_transparent_image(text, font_size=40, padding=0, text_color=(0, 0, 0)):
+    """
+    将文本转换为带透明背景的图像，图像大小刚好包含文本
+    """
+    # 创建临时图像来计算文本尺寸
+    temp_image = Image.new('RGB', (1, 1), (255, 255, 255))
+    temp_draw = ImageDraw.Draw(temp_image)
+    
+    font = ImageFont.truetype(FONTS_PATH / "montserrat.otf", font_size)
+    
+    # 获取文本边界框
+    bbox = temp_draw.textbbox((0, 0), text, font=font)
+    text_width = bbox[2] - bbox[0]
+    text_height = bbox[3] - bbox[1]
+    
+    # 计算图像大小（文本大小 + 内边距）
+    image_width = int(text_width + 2 * padding)
+    image_height = int(text_height + 2 * padding)
+    
+    # 创建RGBA模式的空白图像（带透明通道）
+    image = Image.new('RGBA', (image_width, image_height), (0, 0, 0, 0))
+    draw = ImageDraw.Draw(image)
+    
+    # 绘制文本（考虑内边距）
+    x = padding - bbox[0]  # 调整起始位置
+    y = padding - bbox[1]
+    
+    # 设置文本颜色（带透明度）
+    if len(text_color) == 3:
+        text_color = text_color + (255,)  # 添加完全不透明的alpha值
+    
+    draw.text((x, y), text, fill=text_color, font=font)
+    
+    # 转换为OpenCV格式（BGRA）
+    image_cv = cv2.cvtColor(np.array(image), cv2.COLOR_RGBA2BGRA)
+    return image_cv
+
+def perspective_transform(image, target, corners):
+    """
+    对图像进行透视变换（保持透明通道）
+    target: 画布
+    corners: 四个角点的坐标，顺序为 [左上, 右上, 右下, 左下]
+    """
+    height, width = image.shape[:2]
+    
+    # 源点（原始图像的四个角）
+    src_points = np.array([
+        [0, 0],           # 左上
+        [width-1, 0],     # 右上
+        [width-1, height-1], # 右下
+        [0, height-1]     # 左下
+    ], dtype=np.float32)
+    
+    # 目标点（变换后的四个角）
+    dst_points = np.array(corners, dtype=np.float32)
+    
+    # 计算透视变换矩阵
+    matrix = cv2.getPerspectiveTransform(src_points, dst_points)
+    
+    # 获取画布大小
+    target_height, target_width = target.shape[:2]
+
+    # 应用透视变换（保持所有通道，包括alpha）
+    transformed = cv2.warpPerspective(image, matrix, (target_width, target_height), flags=cv2.INTER_LINEAR)
+
+    return transformed, matrix
+
+def blend_with_transparency(background, foreground, position):
+    """
+    将带透明通道的前景图像合成到背景图像上
+    position: 前景图像在背景图像上的位置 (x, y)
+    """
+    bg = background.copy()
+    
+    # 如果背景没有alpha通道，添加一个
+    if bg.shape[2] == 3:
+        bg = cv2.cvtColor(bg, cv2.COLOR_BGR2BGRA)
+        bg[:, :, 3] = 255  # 完全不透明
+    
+    x, y = position
+    fg_height, fg_width = foreground.shape[:2]
+    bg_height, bg_width = bg.shape[:2]
+    
+    # 确保位置在图像范围内
+    x = max(0, min(x, bg_width - fg_width))
+    y = max(0, min(y, bg_height - fg_height))
+    
+    # 提取前景的alpha通道并归一化
+    alpha_foreground = foreground[:, :, 3] / 255.0
+    
+    # 对于每个颜色通道进行合成
+    for c in range(3):
+        bg_region = bg[y:y+fg_height, x:x+fg_width, c]
+        fg_region = foreground[:, :, c]
+        
+        # alpha混合公式
+        bg[y:y+fg_height, x:x+fg_width, c] = (
+            alpha_foreground * fg_region + 
+            (1 - alpha_foreground) * bg_region
+        )
+    
+    # 更新背景的alpha通道（如果需要）
+    bg_alpha_region = bg[y:y+fg_height, x:x+fg_width, 3]
+    bg[y:y+fg_height, x:x+fg_width, 3] = np.maximum(bg_alpha_region, foreground[:, :, 3])
+    
+    return bg
+
+def precise_blend_with_perspective(background, foreground, corners):
+    """
+    精确合成：根据四个角点将前景图像透视合成到背景上
+    """
+    # 创建与背景相同大小的空白图像
+    bg_height, bg_width = background.shape[:2]
+    
+    # 如果背景没有alpha通道，转换为BGRA
+    if background.shape[2] == 3:
+        background_bgra = cv2.cvtColor(background, cv2.COLOR_BGR2BGRA)
+    else:
+        background_bgra = background.copy()
+    
+    # 创建与背景相同大小的前景图层
+    foreground_layer = np.zeros((bg_height, bg_width, 4), dtype=np.uint8)
+    
+    # 计算前景图像在背景中的边界框
+    min_x = int(min(corners[:, 0]))
+    max_x = int(max(corners[:, 0]))
+    min_y = int(min(corners[:, 1]))
+    max_y = int(max(corners[:, 1]))
+    
+    # 将变换后的前景图像放置到对应位置
+    fg_height, fg_width = foreground.shape[:2]
+    if min_y + fg_height <= bg_height and min_x + fg_width <= bg_width:
+        foreground_layer[min_y:min_y+fg_height, min_x:min_x+fg_width] = foreground
+    
+    # 创建掩码（只在前景有内容的地方合成）
+    mask = (foreground_layer[:, :, 3] > 0)
+    
+    # 合成图像
+    result = background_bgra.copy()
+    for c in range(3):
+        result[:, :, c][mask] = foreground_layer[:, :, c][mask]
+    result[:, :, 3][mask] = foreground_layer[:, :, 3][mask]
+    
+    return result
+
+def wiggle_transform(image, intensity=2) -> list[np.ndarray]:
+    '''
+    返回一组图像振动的帧组，模拟空调运作时的抖动效果
+    '''
+    frames = []
+    height, width = image.shape[:2]
+    shifts = [(-intensity, 0), (intensity, 0), (0, -intensity), (0, intensity), (0, 0)]
+    for dx, dy in shifts:
+        M = np.float32([[1, 0, dx], [0, 1, dy]])
+        shifted = cv2.warpAffine(image, M, (width, height))
+        frames.append(shifted)
+    return frames
+
+async def generate_ac_image(ac: AirConditioner) -> BytesIO:
+    # 找到空调底图
+    ac_image = cv2.imread(str(ASSETS_PATH / "img" / "ac" / "ac.png"), cv2.IMREAD_UNCHANGED)
+
+    if not ac.on:
+        # 空调关闭状态，直接返回底图
+        pil_final = Image.fromarray(ac_image)
+        output = BytesIO()
+        pil_final.save(output, format="GIF")
+        return output
+
+    # 根据生成温度文本图像
+    text = f"{round(ac.temperature, 1)}°C"
+    text_image = text_to_transparent_image(
+        text, 
+        font_size=60, 
+        text_color=(0, 0, 0)  # 黑色文字
+    )
+
+    # 获取长宽比
+    height, width = text_image.shape[:2]
+    aspect_ratio = width / height
+
+    # 定义3D变换的四个角点（透视效果）
+    # 顺序: [左上, 右上, 右下, 左下]
+    corners = np.array([
+        [123, 45],    # 左上
+        [284, 101],   # 右上
+        [290, 140],   # 右下
+        [119, 100]    # 左下
+    ], dtype=np.float32)
+    
+    # 对文本图像进行3D变换（保持透明通道）
+    transformed_text, transform_matrix = perspective_transform(text_image, ac_image, corners)
+
+    final_image_simple = blend_with_transparency(ac_image, transformed_text, (0, 0))
+
+    intensity = max(2, abs(int(ac.temperature) - 24) // 2)
+
+    frames = wiggle_transform(final_image_simple, intensity=intensity)
+    pil_frames = [Image.fromarray(frame) for frame in frames]
+    output = BytesIO()
+    pil_frames[0].save(output, format="GIF", save_all=True, append_images=pil_frames[1:], loop=0, duration=50, disposal=2)
+    return output

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/errman.py

@@ -0,0 +1,45 @@
+from typing import Any
+
+from nonebot.adapters import Bot
+from nonebot.matcher import Matcher
+from nonebot.message import run_postprocessor
+from nonebot_plugin_alconna import UniMessage
+from returns.primitives.exceptions import UnwrapFailedError
+
+from konabot.common.nb.exc import BotExceptionMessage
+
+
+@run_postprocessor
+async def _(bot: Bot, matcher: Matcher, exc: BotExceptionMessage | AssertionError | UnwrapFailedError):
+    if isinstance(exc, BotExceptionMessage):
+        msg = exc.msg
+        await matcher.send(await msg.export(bot))
+    if isinstance(exc, AssertionError):
+        if exc.args:
+            err_msg = exc.args[0]
+
+            err_msg_res: UniMessage
+            if isinstance(err_msg, str):
+                err_msg_res = UniMessage().text(err_msg)
+            elif isinstance(err_msg, UniMessage):
+                err_msg_res = err_msg
+            else:
+                return
+
+            await matcher.send(await err_msg_res.export(bot))
+    if isinstance(exc, UnwrapFailedError):
+        obj = exc.halted_container
+        try:
+            failure: Any = obj.failure()
+
+            err_msg_res: UniMessage
+            if isinstance(failure, str):
+                err_msg_res = UniMessage().text(failure)
+            elif isinstance(failure, UniMessage):
+                err_msg_res = failure
+            else:
+                return
+
+            await matcher.send(await err_msg_res.export(bot))
+        except:
+            pass

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/hanzi/__init__.py

@@ -0,0 +1,217 @@
+import random
+from typing import Optional
+import opencc
+
+from nonebot import on_message
+from nonebot.adapters import Event as BaseEvent
+from nonebot.adapters.discord.event import MessageEvent as DiscordMessageEvent
+from nonebot_plugin_alconna import (
+    Alconna,
+    Args,
+    UniMessage,
+    UniMsg,
+    on_alconna,
+)
+
+convert_type = ["简","簡","繁","正","港","日"]
+
+compiled_str = "|".join([f"{a}{mid}{b}" for mid in ["转","轉","転"] for a in convert_type for b in convert_type if a != b])
+
+def hanzi_to_abbr(hanzi: str) -> str:
+    mapping = {
+        "简": "s",
+        "簡": "s",
+        "繁": "t",
+        "正": "t",
+        "港": "hk",
+        "日": "jp",
+    }
+    return mapping.get(hanzi, "")
+
+def check_valid_convert_type(convert_type: str) -> bool:
+    avaliable_set = ["s2t","t2s","s2tw","tw2s","s2hk","hk2s","s2twp","tw2sp","t2tw","hk2t","t2hk","t2jp","jp2t","tw2t"]
+    if convert_type in avaliable_set:
+        return True
+    return False
+
+def convert(source, src_abbr, dst_abbr):
+    convert_type_key = f"{src_abbr}2{dst_abbr}"
+    if not check_valid_convert_type(convert_type_key):
+        # 先转为繁体，再转为目标
+        converter = opencc.OpenCC(f"{src_abbr}2t.json")
+        source = converter.convert(source)
+        src_abbr = "t"
+    converter = opencc.OpenCC(f"{src_abbr}2{dst_abbr}.json")
+    converted = converter.convert(source)
+    return converted
+
+evt = on_alconna(
+    Alconna(
+        f"re:({compiled_str})",
+        Args["source?", str],
+    ),
+    use_cmd_start=True,
+    use_cmd_sep=False,
+    skip_for_unmatch=True,
+)
+
+@evt.handle()
+async def _(msg: UniMsg, event: BaseEvent, source: Optional[str] = None):
+    if isinstance(event, DiscordMessageEvent):
+        content = event.get_message().extract_plain_text()
+    else:
+        content = event.get_message().extract_plain_text()
+
+    prefix = content.split()[0]
+    to_convert = ""
+    # 如果回复了消息，则转换回复的内容
+    if(source is None):
+        if event.reply:
+            to_convert = event.reply.message.extract_plain_text()
+            if not to_convert:
+                return
+        else:
+            return
+    else:
+        to_convert = source
+    parts = []
+    if "转" in prefix:
+        parts = prefix.split("转")
+    elif "轉" in prefix:
+        parts = prefix.split("轉")
+    elif "転" in prefix:
+        parts = prefix.split("転")
+    if len(parts) != 2:
+        notice = "转换格式错误，请使用“简转繁”、“繁转简”等格式。"
+        await evt.send(await UniMessage().text(notice).export())
+        return
+    src, dst = parts
+    src_abbr = hanzi_to_abbr(src)
+    dst_abbr = hanzi_to_abbr(dst)
+    if not src_abbr or not dst_abbr:
+        notice = "不支持的转换类型，请使用“简”、“繁”、“正”、“港”、“日”等。"
+        if src_abbr:
+            notice = convert(notice, "s", src_abbr)
+        await evt.send(await UniMessage().text(notice).export())
+        return
+    
+    converted = convert(to_convert, src_abbr, dst_abbr)
+
+    converted_prefix = convert("转换结果", "s", dst_abbr)
+
+    await evt.send(await UniMessage().text(f"{converted_prefix}：{converted}").export())
+
+shuo = ["说","說"]
+
+full_name_type = ["简体","簡體","繁體","繁体","正體","正体","港話","港话","日文"]
+
+combined_list = [f"{a}{b}" for a in shuo for b in full_name_type]
+
+compiled_str_2 = "|".join(combined_list)
+
+evt = on_alconna(
+    Alconna(
+        f"re:({compiled_str_2})",
+        Args["source?", str]
+    ),
+    use_cmd_start=True,
+    use_cmd_sep=False,
+    skip_for_unmatch=True,
+)
+
+@evt.handle()
+async def _(msg: UniMsg, event: BaseEvent, source: Optional[str] = None):
+    if isinstance(event, DiscordMessageEvent):
+        content = event.get_message().extract_plain_text()
+    else:
+        content = event.get_message().extract_plain_text()
+
+    prefix = content.split()[0]
+    to_convert = ""
+    # 如果回复了消息，则转换回复的内容
+    if(source is None):
+        if event.reply:
+            to_convert = event.reply.message.extract_plain_text()
+            if not to_convert:
+                return
+        else:
+            return
+    else:
+        to_convert = source
+    # 获取目标转换类型
+    dst = ""
+    match prefix:
+        case "说简体" | "說簡體" | "说簡體" | "說简体":
+            dst = "简"
+        case "說繁體" | "说繁体" | "說繁体" | "说繁體":
+            dst = "繁"
+        case "說正體" | "说正体" | "說正体" | "说正體":
+            dst = "正"
+        case "說港話" | "说港话" | "說港话" | "说港話":
+            dst = "港"
+        case "說日文" | "说日文":
+            dst = "日"
+    dst_abbr = hanzi_to_abbr(dst)
+    if not dst_abbr:
+        notice = "不支持的转换类型，请使用“简体”、“繁體”、“正體”、“港話”、“日文”等。"
+        await evt.send(await UniMessage().text(notice).export())
+        return
+    # 循环，将源语言一次次转换为目标语言
+    current_text = to_convert
+    for src_abbr in ["s","hk","jp","tw","t"]:
+        if src_abbr != dst_abbr:
+            current_text = convert(current_text, src_abbr, dst_abbr)
+
+    converted_prefix = convert("转换结果", "s", dst_abbr)
+
+    await evt.send(await UniMessage().text(f"{converted_prefix}：{current_text}").export())
+
+def random_char(char: str) -> str:
+    dst_abbr = random.choice(["s","t","hk","jp","tw"])
+    for src_abbr in ["s","hk","jp","tw","t"]:
+        if src_abbr != dst_abbr:
+            char = convert(char, src_abbr, dst_abbr)
+    return char
+
+def random_string(text: str) -> str:
+    final_text = ""
+    for char in text:
+        final_text += random_char(char)
+    return final_text
+
+random_match = ["混乱字形","混亂字形","乱数字形","亂數字形","ランダム字形"]
+
+evt = on_alconna(
+    Alconna(
+        f"re:({'|'.join(random_match)})",
+        Args["source?", str]
+    ),
+    use_cmd_start=True,
+    use_cmd_sep=False,
+    skip_for_unmatch=True,
+)
+@evt.handle()
+async def _(msg: UniMsg, event: BaseEvent, source: Optional[str] = None):
+    if isinstance(event, DiscordMessageEvent):
+        content = event.get_message().extract_plain_text()
+    else:
+        content = event.get_message().extract_plain_text()
+
+    prefix = content.split()[0]
+    to_convert = ""
+    # 如果回复了消息，则转换回复的内容
+    if(source is None):
+        if event.reply:
+            to_convert = event.reply.message.extract_plain_text()
+            if not to_convert:
+                return
+        else:
+            return
+    else:
+        to_convert = source
+
+    final_text = ""
+    final_text = random_string(to_convert)
+    converted_prefix = convert(random_string("转换结果"), "s", "s")
+
+    await evt.send(await UniMessage().text(f"{converted_prefix}：{final_text}").export())

konabot/plugins/idiomgame/__init__.py

@@ -0,0 +1,630 @@
+import asyncio as asynkio
+import datetime
+import json
+import secrets
+from enum import Enum
+from pathlib import Path
+from typing import Optional
+
+from loguru import logger
+from nonebot import on_message
+from nonebot.adapters import Event as BaseEvent
+from nonebot.adapters.discord.event import MessageEvent as DiscordMessageEvent
+from nonebot_plugin_alconna import (
+    Alconna,
+    Args,
+    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
+    VERIFIED_AND_REAL = 1
+    ALREADY_USED = 2
+    NOT_IDIOM = 3
+    WRONG_FIRST_CHAR = 4
+    BUT_NO_NEXT = 5
+    GAME_END = 6
+
+
+class IdiomGame:
+    ALL_WORDS = []  # 所有四字词语
+    ALL_IDIOMS = []  # 所有成语
+    INSTANCE_LIST: dict[str, "IdiomGame"] = {}  # 群号对应的游戏实例
+    IDIOM_FIRST_CHAR = {}  # 所有成语包括词语的首字字典
+    AVALIABLE_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  # 剩余回合数
+        self.already_idioms: dict[str, int] = {} # 已经使用过的成语和使用过的次数
+        self.idiom_history: list[list[str]] = [] # 成语使用历史记录，多个数组以存储不同成语链
+        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 = 3
+        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()
+        else:
+            self.add_history_idiom(self.last_idiom, new_chain=True)
+        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 = ""
+        self.all_buff_score = 0
+        self.already_idioms = {}
+        self.idiom_history = []
+
+    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:
+        """
+        跳过当前成语，选择下一个成语
+        """
+        async with self.lock:
+            self._skip_idiom_async()
+            self.add_buff_score(buff_score)
+        return self.last_idiom
+
+    def _skip_idiom_async(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._skip_idiom_async()
+        else:
+            self.add_history_idiom(self.last_idiom, new_chain=True)
+        return self.last_idiom
+
+    async def try_verify_idiom(self, idiom: str, user_id: str) -> TryVerifyState:
+        """
+        用户发送成语
+        """
+        async with self.lock:
+            state = self._verify_idiom(idiom, user_id)
+        return state
+
+    def is_nextable(self, last_char: str) -> bool:
+        """
+        判断是否有成语可以接
+        """
+        return last_char in IdiomGame.AVALIABLE_IDIOM_FIRST_CHAR
+    
+    def add_already_idiom(self, idiom: str):
+        if idiom in self.already_idioms:
+            self.already_idioms[idiom] += 1
+        else:
+            self.already_idioms[idiom] = 1
+
+    def get_already_used_num(self, idiom: str) -> int:
+        if idiom in self.already_idioms:
+            return self.already_idioms[idiom]
+        return 0
+        
+    def add_history_idiom(self, idiom: str, new_chain: bool = False):
+        if new_chain or len(self.idiom_history) == 0:
+            self.idiom_history.append([idiom])
+        else:
+            self.idiom_history[-1].append(idiom)
+
+    def display_history(self) -> list[str]:
+        result = []
+        for chain in self.idiom_history:
+            result.append(" -> ".join(chain))
+        return result
+
+    def _verify_idiom(self, idiom: str, user_id: str) -> list[TryVerifyState]:
+        state = []
+        # 新成语的首字应与上一条成语的尾字相同
+        if idiom[0] != self.last_char:
+            state.append(TryVerifyState.WRONG_FIRST_CHAR)
+            return state
+        if idiom not in IdiomGame.ALL_IDIOMS and idiom not in IdiomGame.ALL_WORDS:
+            self.add_score(user_id, -0.1)
+            state.append(TryVerifyState.NOT_IDIOM)
+            return state
+        # 成语合法，更新状态
+        self.add_history_idiom(idiom)
+        score_k = 0.5 ** self.get_already_used_num(idiom) # 每被使用过一次，得分减半
+        if(score_k != 1):
+            state.append(TryVerifyState.ALREADY_USED)
+        self.add_already_idiom(idiom)
+        state.append(TryVerifyState.VERIFIED)
+        self.last_idiom = idiom
+        self.last_char = idiom[-1]
+        self.add_score(user_id, 1 * score_k) # 先加 1 分
+        if idiom in IdiomGame.ALL_IDIOMS:
+            state.append(TryVerifyState.VERIFIED_AND_REAL)
+            self.add_score(user_id, 4 * score_k) # 再加 4 分
+        self.remain_rounds -= 1
+        if self.remain_rounds <= 0:
+            self.now_playing = False
+            state.append(TryVerifyState.GAME_END)
+        if not self.is_nextable(self.last_char):
+            # 没有成语可以接了，自动跳过
+            self._skip_idiom_async()
+            self.add_buff_score(-100)
+            state.append(TryVerifyState.BUT_NO_NEXT)
+        return state
+
+    def get_user_score(self, user_id: str) -> float:
+        if user_id not in self.score_board:
+            return 0
+        # 避免浮点数精度问题导致过长
+        handled_score = round(self.score_board[user_id]["score"] + self.all_buff_score, 1)
+        return handled_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 random_idiom_starting_with(cls, first_char: str) -> Optional[str]:
+        cls.init_lexicon()
+        if first_char not in cls.AVALIABLE_IDIOM_FIRST_CHAR:
+            return None
+        return secrets.choice(cls.AVALIABLE_IDIOM_FIRST_CHAR[first_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:
+            jsonData = json.load(f)
+            cls.ALL_WORDS = [item["ci"] for item in jsonData]
+            logger.debug(f"Loaded {len(cls.ALL_WORDS)} words from ci.json")
+            logger.debug(f"Sample words: {cls.ALL_WORDS[:5]}")
+
+        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)
+        logger.debug(f"Loaded {len(COMMON_WORDS)} common words from common.txt")
+        logger.debug(f"Sample common words: {COMMON_WORDS[:5]}")
+
+        # 读取 THUOCL 成语库
+        with open(
+            ASSETS_PATH / "lexicon" / "THUOCL" / "data" / "THUOCL_chengyu.txt",
+            "r",
+            encoding="utf-8",
+        ) as f:
+            THUOCL_IDIOMS = [line.split(" ")[0].split("\t")[0].strip() for line in f]
+            logger.debug(f"Loaded {len(THUOCL_IDIOMS)} idioms from THUOCL_chengyu.txt")
+            logger.debug(f"Sample idioms: {THUOCL_IDIOMS[:5]}")
+
+        # 读取 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].split("\t")[0].strip()
+                        if len(word) == 4:
+                            THUOCL_WORDS.append(word)
+        logger.debug(f"Loaded {len(THUOCL_WORDS)} words from THUOCL txt files")
+        logger.debug(f"Sample words: {THUOCL_WORDS[:5]}")
+
+        # 只有成语的大表
+        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)
+
+        # 根据真正的成语大表，划分出有效成语首字字典
+        for idiom in cls.ALL_IDIOMS:
+            if idiom[0] not in cls.AVALIABLE_IDIOM_FIRST_CHAR:
+                cls.AVALIABLE_IDIOM_FIRST_CHAR[idiom[0]] = []
+            cls.AVALIABLE_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, target: DepLongTaskTarget, rounds: Optional[int] = 100
+):
+    await play_game(event, target, 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 += "无人得分！\n"
+    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": {round(info['score'] + instance.get_all_buff_score(), 1)} 分\n"
+            )
+    if len(instance.idiom_history) == 0:
+        result_text += "\n本局没有任何接龙记录。"
+    else:
+        result_text += "\n你们的接龙记录是：\n"
+        history_lines = instance.display_history()
+        for line in history_lines:
+            result_text += line + "\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
+    avaliable_idiom = IdiomGame.random_idiom_starting_with(instance.get_last_char())
+    # 发送哈哈狗图片
+    with open(ASSETS_PATH / "img" / "dog" / "haha_dog.jpg", "rb") as f:
+        img_data = f.read()
+    await evt.send(await UniMessage().image(raw=img_data).export())
+    await evt.send(await UniMessage().text(f"你们太菜了，全部扣100分！明明还可以接「{avaliable_idiom}」的！").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 TryVerifyState.WRONG_FIRST_CHAR in state:
+        return
+    if TryVerifyState.NOT_IDIOM in state:
+        await evt.send(
+            await UniMessage()
+            .at(user_id)
+            .text(" 接不上！这个不一样！你被扣了 0.1 分！")
+            .export()
+        )
+        return
+    already_used_num = instance.get_already_used_num(user_idiom)
+    if TryVerifyState.VERIFIED_AND_REAL in state:
+        score = 5 * (0.5 ** (already_used_num - 1))
+        if already_used_num > 1:
+            await evt.send(
+                await UniMessage()
+                .at(user_id)
+                .text(f" 接上了，这是个被重复用过的成语，喜提 {score} 分！你有 {instance.get_user_score(user_id)} 分！")
+                .export()
+            )
+        else:
+            await evt.send(
+                await UniMessage()
+                .at(user_id)
+                .text(f" 接上了，这是个真实成语，喜提 5 分！你有 {instance.get_user_score(user_id)} 分！")
+                .export()
+            )
+    elif TryVerifyState.VERIFIED in state:
+        score = 1 * (0.5 ** (already_used_num - 1))
+        if already_used_num > 1:
+            await evt.send(
+                await UniMessage()
+                .at(user_id)
+                .text(f" 接上了，但重复了，喜提 {score} 分！你有 {instance.get_user_score(user_id)} 分！")
+                .export()
+            )
+        else:
+            await evt.send(
+                await UniMessage()
+                .at(user_id)
+                .text(f" 接上了，喜提 1 分！你有 {instance.get_user_score(user_id)} 分！")
+                .export()
+            )
+    if TryVerifyState.GAME_END in state:
+        await evt.send(await UniMessage().text("全部回合结束！").export())
+        await end_game(event, group_id)
+        return
+    if TryVerifyState.BUT_NO_NEXT in state:
+        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())