Merge pull request 'fix: use Query to properly handle --pingpong flag' (#65) from pi-agent/konabot:fix/giftool-pingpong-bool into master

Reviewed-on: #65

.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,27 @@ 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
+      - docker run --rm gitea.service.jazzwhom.top/mttu-developers/konabot:nightly-${DRONE_COMMIT_SHA} python scripts/test_playwright.py
+      - docker run --rm gitea.service.jazzwhom.top/mttu-developers/konabot:nightly-${DRONE_COMMIT_SHA} python -m pytest --cov=./konabot/ --cov-report term-missing:skip-covered
+  - 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 +67,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 +87,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.dev

@@ -1,3 +1,2 @@
 DRIVER=~fastapi+~httpx+~websockets
-COMMAND_START=["!", "！", "", "/"]
-ALCONNA_USE_ORIGIN=true
+COMMAND_START=["!", "！", "", "/"]

.env.example

@@ -1,4 +1,4 @@
 ENVIRONMENT=dev
 PORT=21333
-
+DATABASE_PATH="./data/database.db"
 ENABLE_CONSOLE=true

.env.prod

@@ -1,3 +1,2 @@
 DRIVER=~fastapi+~httpx+~websockets
-COMMAND_START=["!", "！", "", "/"]
-ALCONNA_USE_ORIGIN=true
+COMMAND_START=["!", "！", "", "/"]

.env.test

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

.gitignore

@@ -1,3 +1,26 @@
+# 基本的数据文件，以及环境用文件
 /.env
+/data
+/pyrightconfig.json
+/pyrightconfig.toml
+/uv.lock
 
-__pycache__
+# 缓存文件
+__pycache__
+/.ruff_cache
+/.pytest_cache
+/.mypy_cache
+/.black_cache
+
+# 可能会偶然生成的 diff 文件
+/*.diff
+
+# 代码覆盖报告
+/.coverage
+/.coverage.db
+/htmlcov
+
+# 对手动创建虚拟环境的人
+/.venv
+/venv
+*.egg-info

.gitmodules

@@ -0,0 +1,6 @@
+[submodule "assets/lexicon/THUOCL"]
+	path = assets/lexicon/THUOCL
+	url = https://github.com/thunlp/THUOCL.git
+[submodule "assets/oracle"]
+	path = assets/oracle
+	url = https://gitea.service.jazzwhom.top/mttu-developers/oracle-source.git

.sqls.yml

@@ -0,0 +1,6 @@
+lowercaseKeywords: false
+connections:
+  - driver: sqlite
+    dataSourceName: "./data/database.db"
+  - driver: sqlite
+    dataSourceName: "./data/perm.sqlite3"

.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/settings.json

@@ -0,0 +1,5 @@
+{
+    "python.REPL.enableREPLSmartSend": false,
+    "python-envs.defaultEnvManager": "ms-python.python:poetry",
+    "python-envs.defaultPackageManager": "ms-python.python:poetry"
+}

.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 并启用自动重载功能"
-        }
-    ]
-}

AGENTS.md

@@ -0,0 +1,188 @@
+# AGENTS.md
+
+本文件面向两类协作者：
+
+- 手写代码的人类朋友
+- 会在此仓库中协助开发的 AI Agents
+
+这个项目以手写为主，欢迎协作，但请先理解这里的约束和结构，再开始改动。
+
+## 项目定位
+
+- 这是一个娱乐性质的、私域使用的 QQ Bot 项目。
+- 虽然主要用于熟人环境，但依然要按“不信任输入”的标准写代码。
+- 不要因为使用场景偏内部，就默认消息内容、安全边界、调用参数一定可靠。
+
+## 基本原则
+
+### 1. 默认不信任用户输入
+
+所有来自聊天消息、命令参数、平台事件等的输入，都应视为不可信。
+
+开发时至少注意以下几点：
+
+- 不假设输入类型正确，先校验再使用。
+- 不假设输入长度合理，注意超长文本、大量参数、异常嵌套结构。
+- 不假设输入内容安全，避免直接拼接到文件路径、SQL、shell 参数、HTML 或模板中。
+- 不假设用户一定按预期使用命令，错误输入要能优雅失败。
+- 对任何外部请求、文件读写、渲染、执行型逻辑，都要先考虑滥用风险。
+
+### 2. 优先保持现有风格
+
+- 这是一个以人工维护为主的项目，改动应尽量贴近现有写法。
+- 除非有明确收益，不要为了“看起来更现代”而大规模重构。
+- 新增能力时，优先复用已有通用模块，而不是重复造轮子。
+
+### 3. 小步修改，影响清晰
+
+- 尽量做局部、明确、可解释的改动。
+- 修改插件时，避免顺手改动无关插件。
+- 如果要调整公共模块，先确认是否会影响大量插件行为。
+
+## 仓库结构
+
+### `konabot/`
+
+核心代码目录。
+
+#### `konabot/common/`
+
+通用模块目录。
+
+- 放置可复用的基础能力、工具模块、公共逻辑。
+- 如果某段逻辑可能被多个插件共享，应优先考虑放到这里。
+- 修改这里的代码时，要额外关注兼容性，因为它可能被很多插件依赖。
+
+#### `konabot/docs/`
+
+Bot 内部文档系统使用的文档目录。
+
+- 这是给用户看的文档来源。
+- 文档会通过 `man` 指令被触发和展示。
+- 虽然文档文件通常使用 `.txt` 后缀，但内容可以按 markdown 风格书写。
+- `.md` 后缀文件会被忽略，因此 `.md` 更适合只留给仓库维护者阅读的附加说明。
+- 文档文件名就是用户查询时使用的指令名，应保持简洁、稳定、易理解。
+
+补充说明：
+
+- `konabot/docs/user/` 是直接面向用户检索的文档。
+- `konabot/docs/lib/` 更偏向维护者参考。
+- `konabot/docs/concepts/` 用于记录概念。
+- `konabot/docs/sys/` 用于特定范围可见的系统文档。
+
+#### `konabot/plugins/`
+
+插件目录。
+
+- 插件数量很多，是本项目最主要的功能承载位置。
+- 插件可以是单文件，也可以是文件夹形式。
+- 新增插件或修改插件时，请先观察相邻插件的组织方式，再决定采用单文件还是目录结构。
+- 如果逻辑已经明显超出单文件可维护范围，应拆成目录插件，不要把一个文件堆得过大。
+
+## 根目录文档
+
+### `docs/`
+
+仓库根目录下的 `docs/` 主要用于记录一些可以通用的模块说明和开发文档。
+
+- 这里的内容主要面向开发和维护。
+- 适合放公共模块说明、集成说明、配置说明、开发笔记。
+- 不要把面向 `man` 指令直接展示给用户的文档放到这里；那类内容应放在 `konabot/docs/` 下。
+
+## 对 AI Agents 的具体要求
+
+如果你是 AI Agent，请遵守以下约定：
+
+### 修改前
+
+- 先阅读将要修改的文件以及相关上下文，不要只凭文件名猜用途。
+- 先判断目标逻辑属于公共模块、用户文档，还是某个具体插件。
+- 如果需求可以在局部完成，就不要扩大改动范围。
+
+### 修改时
+
+- 优先延续现有命名、目录结构和编码风格。
+- 不要因为“顺手”而批量格式化整个项目。
+- 不要擅自重命名大量文件、移动目录、替换现有架构。
+- 涉及用户输入、路径、网络、数据库、渲染时，主动补上必要的校验与防御。
+- 如果要新增 `konabot/common/` 或其他会被多处依赖的模块，优先考虑 NoneBot2 框架下的依赖注入方式，而不是把全局状态或硬编码依赖散落到调用方。
+- 写文档时，区分清楚是给 `man` 系统看的，还是给仓库维护者看的。
+
+### 修改后
+
+- 检查改动是否误伤其他插件或公共模块。
+- 如果新增了用户可见功能，考虑是否需要补充 `konabot/docs/` 下对应文档。
+- 如果新增或调整了通用能力，考虑是否需要补充根目录 `docs/` 下的说明。
+
+## 插件开发建议
+
+- 单个插件内部优先保持自洽，不要把特定业务逻辑过早抽成公共模块。
+- 当多个插件开始重复同类逻辑时，再考虑上移到 `konabot/common/`。
+- 插件应尽量对异常输入有稳定反馈，而不是直接抛出难理解的错误。
+- 如果插件会访问外部服务，要考虑超时、失败降级和返回内容校验。
+
+### 最基本的用户交互书写建议
+
+- 先用清晰、可收敛的规则匹配消息，再进入处理逻辑，不要一上来就在 handler 里兜底解析所有输入。
+- 在 handler 里尽早提取纯文本、拆分命令和参数，并对缺失参数、非法参数、异常格式给出稳定反馈。
+- 如果用户输入只允许有限枚举值，先定义允许集合，再进行归一化和校验。
+- 输出优先保持简单直接；能一句话说明问题时，不要返回难懂的异常堆栈或过度技术化提示。
+- 涉及渲染、网络请求、图片生成等较重操作时，先确认输入合理，再执行昂贵逻辑。
+- 如果插件只是做单一交互，优先保持 handler 简短，把渲染、请求、转换等逻辑拆成独立函数。
+- 倾向于使用 `UniMessage` / `UniMsg` 这一套消息抽象来组织收发消息，而不是把平台细节和文本拼接散落在各处。
+- 倾向于显式构造返回消息并发送，而不是大量依赖 NoneBot2 原生的 `.finish()` 作为主要输出路径，除非该场景确实更简单清晰。
+
+### 关于公共能力的依赖方式
+
+- 新建通用能力时，优先设计成可注入、可替换、可测试的接口。
+- 如果一个模块未来可能被多个插件依赖，优先考虑 NoneBot2 的依赖注入，而不是让调用方手动维护重复的初始化流程。
+- 除非确有必要，不要让插件直接依赖隐藏的全局副作用。
+- 如果使用单例、缓存或全局管理器，要明确其生命周期、并发行为以及关闭时机。
+
+## 运行环境与部署限制
+
+这个项目默认会跑在 Docker 环境里，修改功能时请先意识到运行环境不是一台“什么都有”的开发机。
+
+### 容器环境
+
+- 运行时基础镜像是 `python:3.13-slim`，不是完整桌面 Linux；很多系统库默认不存在。
+- 项目运行依赖 Playwright Chromium、字体库、图形相关库，以及部分额外二进制工具。
+- 构建阶段和运行阶段是分离的；不要假设在 builder 里装过的系统包，runtime 里也一定可用。
+- 额外制品目前通过多阶段构建放进镜像，例如 `typst`。
+
+### Docker 相关要求
+
+- 如果你新增的 Python 依赖背后还需要 Linux 动态库、字体、图形库、编译工具或其他系统包，必须同步检查并在 `Dockerfile` 中补齐。
+- 不要只让本地虚拟环境能跑；要默认以容器可运行作为完成标准之一。
+- 如果新功能依赖系统命令、共享库、浏览器能力或字体，请在提交说明里明确写出原因。
+- `.dockerignore` 当前会排除 `/.env`、`/.git`、`/data` 等内容；不要依赖这些文件被复制进镜像。
+- 关于额外制品的管理，优先先阅读根目录文档 `docs/artifact.md`；适合统一管理的二进制或外部资源，倾向于复用 `konabot/common/artifact.py`，而不是在各插件里各自处理下载和校验。
+
+### 本地运行
+
+- 本地开发可参考 `justfile`，当前主要入口是 `just watch`。
+- 如果你的改动影响启动方式、依赖准备方式或运行命令，记得同步更新对应文档或脚本。
+
+## 分支与协作流程
+
+- 本项目托管在个人 Gitea 实例：`https://gitea.service.jazzwhom.top/mttu-developers/konabot`。
+- 如果需要创建 Pull Request，优先倾向使用 `tea` CLI：`https://gitea.com/gitea/tea`。
+- Pull Request 创建后，当前主要会有自动机器人做初步评审，项目维护者会手动查看；不要催促立即合并，也不要默认会马上进主分支。
+- 如果当前是在仓库本体上直接开发、而不是在 fork 上工作，尽量提醒用户不要直接在主分支持续改动，优先使用功能分支。
+- 除非用户明确要求，否则不要擅自把改动直接合并到主分支。
+
+## 文档编写建议
+
+### 面向 `man` 的文档
+
+- 放在 `konabot/docs/` 对应子目录。
+- 文件名直接对应用户查询名称。
+- 建议内容简洁，优先说明“做什么、怎么用、示例、注意事项”。
+- 使用 `.txt` 后缀；内容可以写成接近 markdown 的可读格式。
+
+### 面向开发者的文档
+
+- 放在仓库根目录 `docs/`。
+- 主要描述公共模块、配置方法、设计说明、维护经验。
+- 可以使用 `.md`。
+

Dockerfile

@@ -1,8 +1,68 @@
-FROM python:3.13-slim
+FROM alpine:latest AS artifacts
+
+RUN apk add --no-cache curl xz
+WORKDIR /tmp
+
+RUN mkdir -p /artifacts
+RUN curl -L -o typst.tar.xz "https://github.com/typst/typst/releases/download/v0.14.2/typst-x86_64-unknown-linux-musl.tar.xz" \
+    && tar -xJf typst.tar.xz \
+    && mv typst-x86_64-unknown-linux-musl/typst /artifacts
+
+RUN chmod -R +x /artifacts/
+
+
+FROM python:3.13-slim AS base
+
+ENV VIRTUAL_ENV=/app/.venv \
+    PATH="/app/.venv/bin:$PATH" \
+    PLAYWRIGHT_BROWSERS_PATH=/usr/lib/pw-browsers
+
+# 安装所有都需要的底层依赖
+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/*
+
+RUN pip install --no-cache-dir uv
 
 WORKDIR /app
-COPY requirements.txt ./
-RUN pip install -r requirements.txt --no-deps
 
-COPY . .
+COPY pyproject.toml poetry.lock ./
+RUN uv sync --no-install-project
+
+
+
+FROM base AS runtime
+
+COPY --from=builder ${VIRTUAL_ENV} ${VIRTUAL_ENV}
+COPY --from=artifacts /artifacts/ /usr/local/bin/
+
+WORKDIR /app
+
+RUN python -m playwright install chromium
+
+COPY bot.py pyproject.toml .env.prod .env.test ./
+COPY assets ./assets
+COPY scripts ./scripts
+COPY konabot ./konabot
+COPY tests ./tests
+
+ENV PYTHONPATH=/app
+
 CMD [ "python", "bot.py" ]

README.md

@@ -1,4 +1,4 @@
-# 此方 bot！
+# konabot
 
 在 MTTU 内部使用的 bot 一只。
 
@@ -63,12 +63,24 @@ code .
 
 配置 `ENABLE_CONSOLE=false`
 
+#### 配置并支持 LLM（大语言模型）
+
+详见[LLM 配置文档](/docs/LLM.md)。
+
+#### 配置 konabot-web 以支持更高级的图片渲染
+
+详见[konabot-web 配置文档](/docs/konabot-web.md)
+
+#### 数据库配置
+
+本项目使用SQLite作为数据库，默认数据库文件位于`./data/database.db`。可以通过设置`DATABASE_PATH`环境变量来指定其他位置。
+
 ### 运行
 
-你可以在 VSCode 的「运行与调试」窗口，启动 `运行 Bot 并调试（自动重载）` 任务来启动 Bot，也可以使用命令行手动启动 Bot：
+使用命令行手动启动 Bot：
 
 ```bash
-poetry run watchfiles bot.main
+poetry run just watch
 ```
 
 如果你不希望自动重载，只是想运行 Bot，可以直接运行：
@@ -83,3 +95,22 @@ poetry run python bot.py
     - [事件响应器](https://nonebot.dev/docs/tutorial/matcher)
     - [事件处理](https://nonebot.dev/docs/tutorial/handler)
     - [Alconna 插件](https://nonebot.dev/docs/best-practice/alconna/)
+
+## 代码测试
+
+本项目使用 pytest 进行自动化测试，你可以把你的测试代码放在 `./tests` 目录下。
+
+使用命令行执行测试：
+
+```bash
+poetry run just test
+```
+
+使用命令行，在浏览器查看测试覆盖率报告：
+
+```bash
+poetry run just coverage
+# 此时会打开一个 :8000 端口的 Web 服务器
+# 你可以在 http://localhost:8000 查看覆盖率报告
+# 在控制台使用 Ctrl+C 关闭这个 Web 服务器
+```

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,13 @@ 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.appcontext import run_afterinit_functions
+from konabot.common.log import init_logger
+from konabot.common.nb.exc import BotExceptionMessage
+from konabot.common.path import LOG_PATH
+from konabot.common.database import get_global_db_manager
+
+
 dotenv.load_dotenv()
 env = os.environ.get("ENVIRONMENT", "prod")
 env_enable_console = os.environ.get("ENABLE_CONSOLE", "none")
@@ -14,12 +21,27 @@ 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()
 
-    if (env != "prod" and env != "test" and env_enable_console.upper() != "FALSE") or (env_enable_console.upper() == "TRUE"):
+    if (env != "prod" and env != "test" and env_enable_console.upper() != "FALSE") or (
+        env_enable_console.upper() == "TRUE"
+    ):
         driver.register_adapter(ConsoleAdapter)
 
     if env_enable_qq.upper() == "TRUE":
@@ -33,8 +55,19 @@ def main():
 
     # nonebot.load_builtin_plugin("echo")
     nonebot.load_plugins("konabot/plugins")
+    nonebot.load_plugin("nonebot_plugin_analysis_bilibili")
+
+    run_afterinit_functions()
+
+    # 注册关闭钩子
+    @driver.on_shutdown
+    async def _():
+        # 关闭全局数据库管理器
+        db_manager = get_global_db_manager()
+        await db_manager.close_all_connections()
 
     nonebot.run()
 
+
 if __name__ == "__main__":
     main()

docs/LLM.md

@@ -0,0 +1,65 @@
+# 大语言模型平台接入
+
+为实现更多神秘小功能，此方 Bot 需要接入 AI。如果你需要参与开发或测试涉及 AI 的相关功能，麻烦请根据下面的文档继续操作。
+
+## 配置项目接入 AI
+
+AI 相关的配置文件在 `data/config/llm.json` 文件中。示例格式如下，这也将是到时候在云端的配置文件格式（给出的模型都会有）：
+
+```json
+{
+  "llms": {
+    "Qwen2.5-7B-Instruct": {
+      "base_url": "https://api.siliconflow.cn/v1",
+      "api_key": "sk-XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX",
+      "model_name": "Qwen/Qwen2.5-7B-Instruct"
+    },
+    "qwen3-max": {
+      "base_url": "https://dashscope.aliyuncs.com/compatible-mode/v1",
+      "api_key": "sk-XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX",
+      "model_name": "qwen3-max"
+    }
+  },
+  "default_llm": "Qwen2.5-7B-Instruct"
+}
+```
+
+其中，形如 `qwen3-max` 的名称，是你在程序中调用 LLM 使用的键名。若不给出，则会默认使用配置文件中指定的默认模型。
+
+```python
+from konabot.common.llm import get_llm
+
+llm = get_llm()               # 获得的是 Qwen2.5-7B-Instruct 模型
+llm = get_llm("qwen3-max")    # 获得的是 qwen3-max 模型
+
+message = await llm.chat([
+  { "role": "system", "content": "你是一只猫娘" },
+  { "role": "user", "content": "晚上好呀！" },
+], timeout=None, max_tokens=16384)
+# 获得了的是 openai.types.chat.ChatCompletionMessage 对象
+
+print(f"AI 返回值：{message.content}")    # 注意 content 可能为 None，需要做空值检测
+
+client = llm.get_openai_client()    # 获得的是一个 OpenAI Client 对象，可以做更多操作
+                                    # 例如，调用 Embedding 模型来做知识库向量化等工作
+```
+
+## 本项目使用的模型清单
+
+为了便利大家使用，我在这里给出该项目将会使用的模型清单，请根据你的开发需求注册并选择你最喜欢的模型。如果需要接入新的模型，或者使用到文档之外的模型，欢迎在这里给出！
+
+### 硅基流动 Qwen/Qwen2.5-7B-Instruct
+
+一个 7B 大小的 AI 模型。其性能不太能指望，但是它小，而且比较快，可以做一些轻量的操作。
+
+该模型是免费的，但是也需要你注册[硅基流动](https://cloud.siliconflow.cn/me/models)账号，并生成 `api_key` 添加到配置文件中。
+
+### 通义千问 qwen3-max
+
+贵但是很先进的最新模型，其能力可以信赖。但是不要拿它做大量工作哦！
+
+在[百炼大模型平台](https://bailian.console.aliyun.com/)注册账号并申请 `api_key`，新用户会赠送 1M tokens，足够做测试了。
+
+## 安全须知
+
+请注意提防 AI 越狱等情况。

docs/artifact.md

@@ -0,0 +1,26 @@
+# artifact 模块说明
+
+`konabot/common/artifact.py` 用于管理项目运行过程中依赖的额外制品，尤其是二进制文件、外部工具和按平台区分的运行时资源。
+
+## 适用场景
+
+- 某个插件或公共模块依赖额外下载的可执行文件或二进制资源。
+- 依赖需要按操作系统或架构区分。
+- 希望在启动时统一检测、按需下载并校验哈希。
+
+如果额外制品适合在镜像构建阶段直接打包进 Docker 镜像，也可以在 `Dockerfile` 中通过多阶段构建处理；但对于需要在运行环境按平台管理、懒下载或统一校验的资源，优先考虑复用 `artifact.py`。
+
+## 推荐做法
+
+- 新增额外制品时，先判断它更适合放进镜像构建阶段，还是更适合交给 `artifact.py` 管理。
+- 如果该资源会被多个插件或环境复用，倾向于统一通过 `ArtifactDepends` 和 `register_artifacts(...)` 管理。
+- 为下载资源提供稳定来源，并填写 `sha256` 校验值，不要只校验“能不能下载下来”。
+- 使用 `required_os` 和 `required_arch` 限制平台，避免无意义下载。
+- 需要代理时，确认其行为与当前 NoneBot2 配置兼容。
+
+## 注意事项
+
+- 不要把是否存在额外制品的判断散落到多个插件里各自实现。
+- 不要跳过哈希校验，除非该资源确实无法提供稳定校验值，并且有明确理由。
+- 如果一个新能力除了额外制品，还依赖 Linux 动态库、字体、浏览器或系统命令，仍然需要同步检查并更新 `Dockerfile`。
+- 如果镜像构建和运行阶段都依赖该制品，要分别确认 builder 和 runtime 的可用性。

docs/database.md

@@ -0,0 +1,223 @@
+# 数据库系统使用文档
+
+本文档详细介绍了本项目中使用的异步数据库系统，包括其架构设计、使用方法和最佳实践。
+
+## 系统概述
+
+本项目的数据库系统基于 `aiosqlite` 库构建，提供了异步的 SQLite 数据库访问接口。系统主要特性包括：
+
+1. **异步操作**：完全支持异步/await模式，适配NoneBot2框架
+2. **连接池**：内置连接池机制，提高数据库访问性能
+3. **参数化查询**：支持安全的参数化查询，防止SQL注入
+4. **SQL文件支持**：可以直接执行SQL文件中的脚本
+5. **类型支持**：支持 `pathlib.Path` 和 `str` 类型的路径参数
+
+## 核心类和方法
+
+### DatabaseManager 类
+
+`DatabaseManager` 是数据库操作的核心类，提供了以下主要方法：
+
+#### 初始化
+```python
+from konabot.common.database import DatabaseManager
+from pathlib import Path
+
+# 使用默认数据库路径
+db = DatabaseManager()
+
+# 指定了义数据库路径
+db = DatabaseManager("./data/myapp.db")
+db = DatabaseManager(Path("./data/myapp.db"))
+```
+
+#### 查询操作
+```python
+# 执行查询语句并返回结果
+results = await db.query("SELECT * FROM users WHERE age > ?", (18,))
+
+# 从SQL文件执行查询
+results = await db.query_by_sql_file("./sql/get_users.sql", (18,))
+```
+
+#### 执行操作
+```python
+# 执行非查询语句
+await db.execute("INSERT INTO users (name, email) VALUES (?, ?)", ("张三", "zhangsan@example.com"))
+
+# 执行SQL脚本（不带参数）
+await db.execute_script("""
+    CREATE TABLE IF NOT EXISTS users (
+        id INTEGER PRIMARY KEY,
+        name TEXT NOT NULL,
+        email TEXT UNIQUE
+    );
+    INSERT INTO users (name, email) VALUES ('测试用户', 'test@example.com');
+""")
+
+# 从SQL文件执行非查询语句
+await db.execute_by_sql_file("./sql/create_tables.sql")
+
+# 带参数执行SQL文件
+await db.execute_by_sql_file("./sql/insert_user.sql", ("张三", "zhangsan@example.com"))
+
+# 执行多条语句（每条语句使用相同参数）
+await db.execute_many("INSERT INTO users (name, email) VALUES (?, ?)", [
+    ("张三", "zhangsan@example.com"),
+    ("李四", "lisi@example.com"),
+    ("王五", "wangwu@example.com")
+])
+
+# 从SQL文件执行多条语句（每条语句使用相同参数）
+await db.execute_many_values_by_sql_file("./sql/batch_insert.sql", [
+    ("张三", "zhangsan@example.com"),
+    ("李四", "lisi@example.com")
+])
+```
+
+## SQL文件处理机制
+
+### 单语句SQL文件
+```sql
+-- insert_user.sql
+INSERT INTO users (name, email) VALUES (?, ?);
+```
+
+```python
+# 使用方式
+await db.execute_by_sql_file("./sql/insert_user.sql", ("张三", "zhangsan@example.com"))
+```
+
+### 多语句SQL文件
+```sql
+-- setup.sql
+CREATE TABLE IF NOT EXISTS users (
+    id INTEGER PRIMARY KEY,
+    name TEXT NOT NULL,
+    email TEXT UNIQUE
+);
+
+CREATE TABLE IF NOT EXISTS profiles (
+    user_id INTEGER,
+    age INTEGER,
+    FOREIGN KEY (user_id) REFERENCES users(id)
+);
+```
+
+```python
+# 使用方式
+await db.execute_by_sql_file("./sql/setup.sql")
+```
+
+### 多语句带不同参数的SQL文件
+```sql
+-- batch_operations.sql
+INSERT INTO users (name, email) VALUES (?, ?);
+INSERT INTO profiles (user_id, age) VALUES (?, ?);
+```
+
+```python
+# 使用方式
+await db.execute_by_sql_file("./sql/batch_operations.sql", [
+    ("张三", "zhangsan@example.com"),  # 第一条语句的参数
+    (1, 25)  # 第二条语句的参数
+])
+```
+
+## 最佳实践
+
+### 1. 数据库表设计
+```sql
+-- 推荐的表设计实践
+CREATE TABLE IF NOT EXISTS example_table (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    name TEXT NOT NULL,
+    created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
+    updated_at DATETIME DEFAULT CURRENT_TIMESTAMP
+);
+```
+
+### 2. SQL文件组织
+建议按照功能模块组织SQL文件：
+```
+plugin/
+├── sql/
+│   ├── create_tables.sql
+│   ├── insert_data.sql
+│   ├── update_data.sql
+│   └── query_data.sql
+└── __init__.py
+```
+
+### 3. 错误处理
+```python
+try:
+    results = await db.query("SELECT * FROM users WHERE id = ?", (user_id,))
+except Exception as e:
+    logger.error(f"数据库查询失败: {e}")
+    # 处理错误情况
+```
+
+### 4. 连接管理
+```python
+# 在应用启动时初始化
+db_manager = DatabaseManager()
+
+# 在应用关闭时清理连接
+async def shutdown():
+    await db_manager.close_all_connections()
+```
+
+## 高级特性
+
+### 连接池配置
+```python
+class DatabaseManager:
+    def __init__(self, db_path: Optional[Union[str, Path]] = None):
+        # 连接池大小配置
+        self._pool_size = 5  # 可根据需要调整
+```
+
+### 事务支持
+```python
+# 通过execute方法的自动提交机制支持事务
+await db.execute("BEGIN TRANSACTION")
+try:
+    await db.execute("INSERT INTO users (name) VALUES (?)", ("张三",))
+    await db.execute("INSERT INTO profiles (user_id, age) VALUES (?, ?)", (1, 25))
+    await db.execute("COMMIT")
+except Exception:
+    await db.execute("ROLLBACK")
+    raise
+```
+
+## 注意事项
+
+1. **异步环境**：所有数据库操作都必须在异步环境中执行
+2. **参数安全**：始终使用参数化查询，避免SQL注入
+3. **资源管理**：确保在应用关闭时调用 `close_all_connections()`
+4. **SQL解析**：使用 `sqlparse` 库准确解析SQL语句，正确处理包含分号的字符串和注释
+5. **错误处理**：适当处理数据库操作可能抛出的异常
+
+## 常见问题
+
+### Q: 如何处理数据库约束错误？
+A: 确保SQL语句中的字段名正确引用，特别是保留字需要使用双引号包围：
+```sql
+CREATE TABLE air_conditioner (
+    id VARCHAR(128) PRIMARY KEY, 
+    "on" BOOLEAN NOT NULL,  -- 使用双引号包围保留字
+    temperature REAL NOT NULL
+);
+```
+
+### Q: 如何处理多个语句和参数的匹配？
+A: 当SQL文件包含多个语句时，参数应该是参数列表，每个语句对应一个参数元组：
+```python
+await db.execute_by_sql_file("./sql/batch.sql", [
+    ("参数1", "参数2"),  # 第一个语句的参数
+    ("参数3", "参数4")   # 第二个语句的参数
+])
+```
+
+通过遵循这些指南和最佳实践，您可以充分利用本项目的异步数据库系统，构建高性能、安全的数据库应用。

docs/konabot-web.md

@@ -0,0 +1,18 @@
+# konabot-web 配置文档
+
+本文档教你配置一个此方 Bot 的 Web 服务器。
+
+## 安装并运行 konabot-web
+
+按照 [konabot-web README](https://gitea.service.jazzwhom.top/mttu-developers/konabot-web) 安装并运行 konabot-web 实例。
+
+## 指定 konabot-web 实例地址
+
+如果你的 Web 服务器的端口不是 5173，或者你有特殊的网络结构，你需要手动设置 konabot-web。编辑 `.env` 文件：
+
+```
+MODULE_WEB_RENDER_WEBURL=http://web-server:port
+MODULE_WEB_RENDER_INSTANCE=http://konabot-server:port
+```
+
+替换 web-server 为你的前端服务器地址，konabot-server 为后端服务器地址，port 为端口号。

docs/permsys.md

@@ -0,0 +1,244 @@
+# 权限系统 `konabot.common.permsys`
+
+本文档面向维护者，说明 `konabot/common/permsys` 模块的职责、数据模型、权限解析规则，以及在插件中接入的推荐方式。
+
+## 模块目标
+
+`permsys` 提供了一套简单的、可继承的权限系统，用于回答两个问题：
+
+1. 某个事件对应的主体是谁。
+2. 该主体是否拥有某项权限。
+
+它适合处理 bot 内部的功能开关、管理权限、平台级授权等场景。
+
+当前模块由以下几部分组成：
+
+- `konabot/common/permsys/__init__.py`
+  - 暴露 `PermManager`、`DepPermManager`、`require_permission`
+  - 负责数据库初始化、启动迁移、超级管理员默认授权
+  - 提供 `register_default_allow_permission()` 用于注册“启动时默认放行”的权限键
+- `konabot/common/permsys/entity.py`
+  - 定义 `PermEntity`
+  - 将事件转换为可查询的实体链
+- `konabot/common/permsys/repo.py`
+  - 封装 SQLite 读写
+- `konabot/common/permsys/migrates/`
+  - 存放迁移 SQL
+- `konabot/common/permsys/sql/`
+  - 存放查询与更新 SQL
+
+## 核心概念
+
+### 1. `PermEntity`
+
+`PermEntity` 是权限系统中的最小主体标识：
+
+```python
+PermEntity(platform: str, entity_type: str, external_id: str)
+```
+
+示例：
+
+- `PermEntity("sys", "global", "global")`
+- `PermEntity("ob11", "group", "123456")`
+- `PermEntity("ob11", "user", "987654")`
+
+其中：
+
+- `platform` 表示来源平台，如 `sys`、`ob11`、`discord`
+- `entity_type` 表示主体类型，如 `global`、`group`、`user`
+- `external_id` 表示平台侧的外部标识
+
+### 2. 实体链
+
+权限判断不是只看单个实体，而是看一条“实体链”。
+
+以 `get_entity_chain_of_entity()` 为例，传入一个具体实体时，返回的链为：
+
+```python
+[
+    PermEntity(platform, entity_type, external_id),
+    PermEntity(platform, "global", "global"),
+    PermEntity("sys", "global", "global"),
+]
+```
+
+这意味着权限会优先读取更具体的主体，再回退到平台全局，最后回退到系统全局。
+
+`get_entity_chain(event)` 则会根据事件类型自动构造链。例如：
+
+- OneBot V11 群消息：用户 -> 群 -> 平台全局 -> 系统全局
+- OneBot V11 私聊：用户 -> 平台全局 -> 系统全局
+- Discord 频道消息：用户/频道/服务器 -> 平台全局 -> 系统全局
+- Console：控制台用户/频道 -> 平台全局 -> 系统全局
+
+注意：当前 `entity.py` 中的具体链顺序与字段命名应以实现为准；修改这里时要评估现有权限继承是否会被破坏。
+
+### 3. 权限键
+
+权限键使用点分结构，例如：
+
+- `admin`
+- `plugin.weather`
+- `plugin.weather.use`
+
+检查时会自动做前缀回退。以 `plugin.weather.use` 为例，查询顺序是：
+
+1. `plugin.weather.use`
+2. `plugin.weather`
+3. `plugin`
+4. `*`
+
+因此，`*` 可以看作兜底总权限。
+
+## 权限解析规则
+
+`PermManager.check_has_permission_info()` 的逻辑可以概括为：
+
+1. 先把输入转换成实体链。
+2. 对权限键做逐级回退，同时追加 `*`。
+3. 在数据库中批量查出链上所有实体、所有候选键的显式记录。
+4. 按“实体越具体越优先、权限键越具体越优先”的顺序，返回第一条命中的记录。
+
+若没有任何显式记录：
+
+- `check_has_permission_info()` 返回 `None`
+- `check_has_permission()` 返回 `False`
+
+这表示本系统默认是“未授权即拒绝”。
+
+## 数据存储
+
+模块使用 SQLite，默认数据库文件位于：
+
+- `data/perm.sqlite3`
+
+启动时会执行迁移：
+
+- `create_startup()` 在 NoneBot 启动事件中调用 `execute_migration()`
+
+权限值支持三态：
+
+- `True`：显式允许
+- `False`：显式拒绝
+- `None`：删除/清空该层的显式设置，让判断重新回退到继承链
+
+`repo.py` 中的 `update_perm_info()` 会将这个三态直接写入数据库。
+
+## 超级管理员注入
+
+在启动阶段，`create_startup()` 会读取 `konabot.common.nb.is_admin.cfg.admin_qq_account`，并为这些 QQ 账号写入：
+
+```python
+PermEntity("ob11", "user", str(account)), "*", True
+```
+
+也就是说，配置中的超级管理员会直接拥有全部权限。
+
+此外，模块也支持插件在导入阶段通过 `register_default_allow_permission("some.key")` 注册默认放行的权限键；这些键会在启动时被写入到：
+
+```python
+PermEntity("sys", "global", "global"), "some.key", True
+```
+
+这适合“默认所有人可用，但仍希望后续能被权限系统单独关闭”的功能。
+
+这属于启动时自动灌入的保底策略，不依赖手工授权命令。
+
+## 在插件中使用
+
+### 1. 直接做权限检查
+
+```python
+from konabot.common.permsys import DepPermManager
+
+
+async def handler(pm: DepPermManager, event):
+    ok = await pm.check_has_permission(event, "plugin.example.use")
+    if not ok:
+        return
+```
+
+适合需要在处理流程中动态决定权限键的场景。
+
+### 2. 挂到 Rule 上做准入控制
+
+```python
+from nonebot_plugin_alconna import Alconna, on_alconna
+from konabot.common.permsys import require_permission
+
+
+cmd = on_alconna(
+    Alconna("example"),
+    rule=require_permission("plugin.example.use"),
+)
+```
+
+适合命令入口明确、未通过时直接拦截的场景。
+
+### 3. 更新权限
+
+```python
+from konabot.common.permsys import DepPermManager
+from konabot.common.permsys.entity import PermEntity
+
+
+await pm.update_permission(
+    PermEntity("ob11", "group", "123456"),
+    "plugin.example.use",
+    True,
+)
+```
+
+建议只在专门的管理插件中开放写权限，避免普通功能插件到处分散改表。
+
+## `perm_manage` 插件与本模块的关系
+
+`konabot/plugins/perm_manage/__init__.py` 是本模块当前的管理入口，提供：
+
+- `konaperm list`：列出实体链上已有的显式权限记录
+- `konaperm get`：查看某个权限最终命中的记录
+- `konaperm set`：写入 allow/deny/null
+
+这个插件本身使用 `require_permission("admin")` 保护，因此只有拥有 `admin` 权限的主体才能管理权限。
+
+## 接入建议
+
+### 权限键命名
+
+建议使用稳定、可扩展的分层键名：
+
+- 推荐：`plugin.xxx`、`plugin.xxx.action`
+- 不推荐：含糊的单词或临时字符串
+
+这样才能利用前缀回退机制做批量授权。
+
+### 输入安全
+
+虽然这个项目偏内部使用，但权限键、实体类型、外部 ID 仍然应视为不可信输入：
+
+- 不要把聊天输入直接拼到 SQL 中
+- 不要让任意用户可随意构造高权限写入
+- 对可写命令至少做权限保护和必要校验
+
+### 改动兼容性
+
+以下改动都可能影响全局权限行为，修改前应充分评估：
+
+- 更改实体链顺序
+- 更改默认兜底键 `*` 的语义
+- 更改 `None` 的处理方式
+- 更改启动时超级管理员注入逻辑
+
+## 调试建议
+
+- 先用 `konaperm get ...` 确认某个权限最终命中了哪一层
+- 再用 `konaperm list ...` 查看该实体链上有哪些显式记录
+- 若表现异常，检查是否是更上层实体或更宽泛权限键提前命中
+
+## 相关文件
+
+- `konabot/common/permsys/__init__.py`
+- `konabot/common/permsys/entity.py`
+- `konabot/common/permsys/repo.py`
+- `konabot/plugins/perm_manage/__init__.py`

docs/subscribe.md

@@ -0,0 +1,37 @@
+# subscribe 模块
+
+一套统一的接口，让用户可以订阅一些延迟或者定时消息。
+
+```python
+import asyncio
+
+from pathlib import Path
+
+from konabot.common.subscribe import register_poster_info, broadcast, PosterInfo
+from nonebot_plugin_alconna import UniMessage
+
+# 注册了服务信息，用户可以用「查询可用订阅」指令了解可用的订阅清单。
+# 用户可以使用「订阅 某某服务通知」或者「订阅 某某服务」来订阅消息。
+# 如果用户在群聊发起订阅，则会在 QQ 群订阅，不然会在私聊订阅
+register_poster_info("某某服务通知", PosterInfo(
+  aliases={"某某服务"},
+  description="告诉你关于某某的最新资讯等信息",
+))
+
+async def main():
+  while True:
+    # 这里的服务 channel 名字必须填写该服务的名字，不可以是 alias
+    # 这会给所有订阅了该通道的用户发送「向大家发送纯文本通知」
+    await broadcast("某某服务通知", "向大家发送纯文本通知")
+
+    # 也可以发送 UniMessage 对象，可以构造包含图片的通知等
+    data = Path('image.png').read_bytes()
+    await broadcast(
+      "某某服务通知",
+      UniMessage.text("很遗憾告诉大家，我们倒闭了：").image(raw=data),
+    )
+
+    await asyncio.sleep(114.514)
+```
+
+该模块的代码请查阅 `/konabot/common/subscribe/` 下的文件。

justfile

@@ -0,0 +1,9 @@
+watch:
+    poetry run watchfiles bot.main . --filter scripts.watch_filter.filter
+
+test:
+    poetry run pytest --cov-report term-missing:skip-covered
+
+coverage:
+    poetry run pytest --cov-report html
+    python -m http.server -d htmlcov

konabot/common/apis/ali_content_safety.py

@@ -0,0 +1,92 @@
+import asyncio
+import json
+
+from alibabacloud_green20220302.client import Client as AlibabaGreenClient
+from alibabacloud_green20220302.models import TextModerationPlusRequest
+from alibabacloud_tea_openapi.models import Config as AlibabaTeaConfig
+from loguru import logger
+from pydantic import BaseModel
+
+import nonebot
+
+
+class AlibabaGreenPluginConfig(BaseModel):
+    module_aligreen_enable: bool = False
+    module_aligreen_access_key_id: str = ""
+    module_aligreen_access_key_secret: str = ""
+    module_aligreen_region_id: str = "cn-shenzhen"
+    module_aligreen_endpoint: str = "green-cip.cn-shenzhen.aliyuncs.com"
+    module_aligreen_service: str = "llm_query_moderation"
+
+
+class AlibabaGreen:
+    _client: AlibabaGreenClient | None = None
+    _config: AlibabaGreenPluginConfig | None = None
+
+    @staticmethod
+    def get_client() -> AlibabaGreenClient:
+        assert AlibabaGreen._client is not None
+        return AlibabaGreen._client
+
+    @staticmethod
+    def get_config() -> AlibabaGreenPluginConfig:
+        assert AlibabaGreen._config is not None
+        return AlibabaGreen._config
+
+    @staticmethod
+    def init():
+        config = nonebot.get_plugin_config(AlibabaGreenPluginConfig)
+        AlibabaGreen._config = config
+        if not config.module_aligreen_enable:
+            logger.info("该环境未启用阿里内容审查，跳过初始化")
+            return
+        AlibabaGreen._client = AlibabaGreenClient(AlibabaTeaConfig(
+            access_key_id=config.module_aligreen_access_key_id,
+            access_key_secret=config.module_aligreen_access_key_secret,
+            connect_timeout=10000,
+            read_timeout=3000,
+            region_id=config.module_aligreen_region_id,
+            endpoint=config.module_aligreen_endpoint,
+        ))
+
+    @staticmethod
+    def _detect_sync(content: str) -> bool:
+        if len(content) == 0:
+            return True
+        if not AlibabaGreen.get_config().module_aligreen_enable:
+            logger.debug("该环境未启用阿里内容审查，直接跳过")
+            return True
+
+        client = AlibabaGreen.get_client()
+        try:
+            response = client.text_moderation_plus(TextModerationPlusRequest(
+                service=AlibabaGreen.get_config().module_aligreen_service,
+                service_parameters=json.dumps({
+                    "content": content,
+                }),
+            ))
+            if response.status_code == 200:
+                result = response.body
+                logger.info(f"检测违规内容 API 调用成功：{result}")
+                risk_level: str = result.data.risk_level or "none"
+                if risk_level == "high":
+                    return False
+                return True
+            logger.error(f"检测违规内容 API 调用失败：{response}")
+            return True
+        except Exception as e:
+            logger.error("检测违规内容 API 调用失败")
+            logger.exception(e)
+            return True
+
+    @staticmethod
+    async def detect(content: str) -> bool:
+        return await asyncio.to_thread(AlibabaGreen._detect_sync, content)
+
+
+driver = nonebot.get_driver()
+
+@driver.on_startup
+async def _():
+    AlibabaGreen.init()
+

konabot/common/apis/wolfx.py

@@ -0,0 +1,281 @@
+"""
+Wolfx 防灾免费 API
+"""
+
+import asyncio
+import json
+from typing import Literal, TypeVar, cast
+import aiohttp
+from aiosignal import Signal
+from loguru import logger
+from pydantic import BaseModel, RootModel
+import pydantic
+
+from konabot.common.appcontext import after_init
+
+
+class ScEewReport(BaseModel):
+    """
+    四川地震局报文
+    """
+
+    ID: str
+    "EEW 发报 ID"
+
+    EventID: str
+    "EEW 发报事件 ID"
+
+    ReportTime: str
+    "EEW 发报时间(UTC+8)"
+
+    ReportNum: int
+    "EEW 发报数"
+
+    OriginTime: str
+    "发震时间(UTC+8)"
+
+    HypoCenter: str
+    "震源地"
+
+    Latitude: float
+    "震源地纬度"
+
+    Longitude: float
+    "震源地经度"
+
+    Magnitude: float
+    "震级"
+
+    Depth: float | None
+    "震源深度"
+
+    MaxIntensity: float
+    "最大烈度"
+
+
+class CencEewReport(BaseModel):
+    """
+    中国地震台网报文
+    """
+
+    ID: str
+    "EEW 发报 ID"
+
+    EventID: str
+    "EEW 发报事件 ID"
+
+    ReportTime: str
+    "EEW 发报时间(UTC+8)"
+
+    ReportNum: int
+    "EEW 发报数"
+
+    OriginTime: str
+    "发震时间(UTC+8)"
+
+    HypoCenter: str
+    "震源地"
+
+    Latitude: float
+    "震源地纬度"
+
+    Longitude: float
+    "震源地经度"
+
+    Magnitude: float
+    "震级"
+
+    Depth: float | None
+    "震源深度"
+
+    MaxIntensity: float
+    "最大烈度"
+
+
+class CencEqReport(BaseModel):
+    type: str
+    "报告类型"
+
+    EventID: str
+    "事件 ID"
+
+    time: str
+    "UTC+8 格式的地震发生时间"
+
+    location: str
+    "地震发生位置"
+
+    magnitude: str
+    "震级"
+
+    depth: str
+    "地震深度"
+
+    latitude: str
+    "纬度"
+
+    longtitude: str
+    "经度"
+
+    intensity: str
+    "烈度"
+
+
+class CencEqlist(RootModel):
+    root: dict[str, CencEqReport]
+
+
+class WolfxWebSocket:
+    def __init__(self, url: str) -> None:
+        self.url = url
+        self.signal: Signal[bytes] = Signal(self)
+        self._running = False
+        self._task: asyncio.Task | None = None
+        self._session: aiohttp.ClientSession | None = None
+        self._ws: aiohttp.ClientWebSocketResponse | None = None
+
+    @property
+    def session(self) -> aiohttp.ClientSession:  # pragma: no cover
+        assert self._session is not None
+        return self._session
+
+    async def start(self):  # pragma: no cover
+        if self._running:
+            return
+        self._running = True
+        self._session = aiohttp.ClientSession()
+        self._task = asyncio.create_task(self._run())
+        self.signal.freeze()
+
+    async def stop(self):  # pragma: no cover
+        self._running = False
+        if self._task:
+            self._task.cancel()
+            try:
+                await self._task
+            except asyncio.CancelledError:
+                pass
+            if self._session:
+                await self._session.close()
+
+    async def _run(self):  # pragma: no cover
+        retry_delay = 1
+
+        while self._running:
+            try:
+                async with self.session.ws_connect(self.url) as ws:
+                    self._ws = ws
+                    logger.info(f"Wolfx API 服务连接上了 {self.url} 的 WebSocket")
+                    async for msg in ws:
+                        if msg.type == aiohttp.WSMsgType.TEXT:
+                            await self.handle(cast(str, msg.data).encode())
+                        elif msg.type == aiohttp.WSMsgType.BINARY:
+                            await self.handle(cast(bytes, msg.data))
+                        elif msg.type == aiohttp.WSMsgType.CLOSED:
+                            break
+                        elif msg.type == aiohttp.WSMsgType.ERROR:
+                            break
+            except (aiohttp.ClientError, asyncio.TimeoutError) as e:
+                logger.warning("连接 WebSocket 时发生错误")
+                logger.exception(e)
+            except asyncio.CancelledError:
+                break
+            except Exception as e:
+                logger.error("Wolfx API 发生未知错误")
+                logger.exception(e)
+            self._ws = None
+
+            if self._running:
+                logger.info(f"Wolfx API 准备断线重连 {self.url}")
+                await asyncio.sleep(retry_delay)
+                retry_delay = min(retry_delay * 2, 60)
+
+    async def handle(self, data: bytes):
+        try:
+            obj = json.loads(data)
+        except json.JSONDecodeError as e:
+            logger.warning("解析 Wolfs API 时出错")
+            logger.exception(e)
+            return
+
+        if obj.get("type") == "heartbeat" or obj.get("type") == "pong":
+            logger.debug(f"Wolfx API 收到了来自 {self.url} 的心跳: {obj}")
+        else:
+            await self.signal.send(data)
+
+
+T = TypeVar("T", bound=BaseModel)
+
+
+class WolfxAPIService:
+    sc_eew: Signal[ScEewReport]
+    "四川地震局地震速报"
+
+    cenc_eew: Signal[CencEewReport]
+    "中国地震台网地震速报"
+
+    cenc_eqlist: Signal[CencEqReport]
+    "中国地震台网地震信息发布"
+
+    def __init__(self) -> None:
+        self.sc_eew = Signal(self)
+        self._sc_eew_ws = WolfxWebSocket("wss://ws-api.wolfx.jp/sc_eew")
+        WolfxAPIService.bind(self.sc_eew, self._sc_eew_ws, ScEewReport)
+
+        self.cenc_eew = Signal(self)
+        self._cenc_eew_ws = WolfxWebSocket("wss://ws-api.wolfx.jp/cenc_eew")
+        WolfxAPIService.bind(self.cenc_eew, self._cenc_eew_ws, CencEewReport)
+
+        self.cenc_eqlist = Signal(self)
+        self._cenc_eqlist_ws = WolfxWebSocket("wss://ws-api.wolfx.jp/cenc_eqlist")
+        WolfxAPIService.bind(self.cenc_eqlist, self._cenc_eqlist_ws, CencEqReport)
+
+    @staticmethod
+    def bind(signal: Signal[T], ws: WolfxWebSocket, t: type[T]):
+        @ws.signal.append
+        async def _(data: bytes):
+            try:
+                obj = t.model_validate_json(data)
+                logger.info(f"接收到来自 Wolfx API 的信息：{data}")
+                await signal.send(obj)
+            except pydantic.ValidationError as e:
+                logger.warning(f"解析 Wolfx API 时出错 URL={ws.url}")
+                logger.error(e)
+
+    async def start(self):  # pragma: no cover
+        self.cenc_eew.freeze()
+        self.sc_eew.freeze()
+        self.cenc_eqlist.freeze()
+        async with asyncio.TaskGroup() as task_group:
+            if len(self.cenc_eew) > 0:
+                task_group.create_task(self._cenc_eew_ws.start())
+
+            if len(self.sc_eew) > 0:
+                task_group.create_task(self._sc_eew_ws.start())
+
+            if len(self.cenc_eqlist) > 0:
+                task_group.create_task(self._cenc_eqlist_ws.start())
+
+    async def stop(self):  # pragma: no cover
+        async with asyncio.TaskGroup() as task_group:
+            task_group.create_task(self._cenc_eew_ws.stop())
+            task_group.create_task(self._sc_eew_ws.stop())
+            task_group.create_task(self._cenc_eqlist_ws.stop())
+
+
+wolfx_api = WolfxAPIService()
+
+
+@after_init
+def init():  # pragma: no cover
+    import nonebot
+
+    driver = nonebot.get_driver()
+
+    @driver.on_startup
+    async def _():
+        await wolfx_api.start()
+
+    @driver.on_shutdown
+    async def _():
+        await wolfx_api.stop()

konabot/common/appcontext.py

@@ -0,0 +1,15 @@
+from typing import Any, Callable
+
+
+AFTER_INIT_FUNCTION = Callable[[], Any]
+
+_after_init_functions: list[AFTER_INIT_FUNCTION] = []
+
+
+def after_init(func: AFTER_INIT_FUNCTION):
+    _after_init_functions.append(func)
+
+
+def run_afterinit_functions():  # pragma: no cover
+    for f in _after_init_functions:
+        f()

konabot/common/artifact.py

@@ -0,0 +1,112 @@
+import asyncio
+import aiohttp
+import hashlib
+import platform
+
+from dataclasses import dataclass
+from pathlib import Path
+
+import nonebot
+from loguru import logger
+from nonebot.adapters.discord.config import Config as DiscordConfig
+from pydantic import BaseModel
+
+
+@dataclass
+class ArtifactDepends:
+    url: str
+    sha256: str
+    target: Path
+
+    required_os: str | None = None
+    "示例值：Windows, Linux, Darwin"
+
+    required_arch: str | None = None
+    "示例值：AMD64, x86_64, arm64"
+
+    use_proxy: bool = True
+    "网络问题，赫赫；使用的是 Discord 模块配置的 proxy"
+
+    def is_corresponding_platform(self) -> bool:
+        if self.required_os is not None:
+            if self.required_os.lower() != platform.system().lower():
+                return False
+        if self.required_arch is not None:
+            if self.required_arch.lower() != platform.machine().lower():
+                return False
+        return True
+
+
+class Config(BaseModel):
+    prefetch_artifact: bool = False
+    "是否提前下载好二进制依赖"
+
+
+artifact_list = []
+
+
+driver = nonebot.get_driver()
+config = nonebot.get_plugin_config(Config)
+
+@driver.on_startup
+async def _():
+    if config.prefetch_artifact:
+        logger.info("启动检测中：正在检测需求的二进制是否下载")
+        semaphore = asyncio.Semaphore(10)
+        async def _task(artifact: ArtifactDepends):
+            async with semaphore:
+                await ensure_artifact(artifact)
+
+        tasks: set[asyncio.Task] = set()
+        for a in artifact_list:
+            tasks.add(asyncio.Task(_task(a)))
+        await asyncio.gather(*tasks, return_exceptions=False)
+        logger.info("检测好了")
+
+
+async def download_artifact(artifact: ArtifactDepends):
+    proxy = None
+    if artifact.use_proxy:
+        discord_config = nonebot.get_plugin_config(DiscordConfig)
+        proxy = discord_config.discord_proxy
+
+    if proxy is not None:
+        logger.info(f"正在使用 Proxy 下载 TARGET={artifact.target} PROXY={proxy}")
+    else:
+        logger.info(f"正在下载 TARGET={artifact.target}")
+
+    async with aiohttp.ClientSession(proxy=proxy) as client:
+        result = await client.get(artifact.url)
+        if result.status != 200:
+            logger.warning(f"已经下载了二进制，但是注意服务器没有返回 200！ URL={artifact.url} TARGET={artifact.target} CODE={result.status}")
+        data = await result.read()
+    artifact.target.write_bytes(data)
+    if not platform.system().lower() == 'windows':
+        artifact.target.chmod(0o755)
+
+    logger.info(f"下载好了 TARGET={artifact.target} URL={artifact.url}")
+    m = hashlib.sha256(artifact.target.read_bytes())
+    if m.hexdigest().lower() != artifact.sha256.lower():
+        logger.warning(f"下载到的二进制的 sha256 与需求不同 TARGET={artifact.target} REQUESTED={artifact.sha256} ACTUAL={m.hexdigest()}")
+
+
+async def ensure_artifact(artifact: ArtifactDepends):
+    if not artifact.is_corresponding_platform():
+        return
+
+    if not artifact.target.exists():
+        logger.info(f"二进制依赖 {artifact.target} 不存在")
+        if not artifact.target.parent.exists():
+            artifact.target.parent.mkdir(parents=True, exist_ok=True)
+        await download_artifact(artifact)
+    else:
+        m = hashlib.sha256(artifact.target.read_bytes())
+        if m.hexdigest().lower() != artifact.sha256.lower():
+            logger.info(f"二进制依赖 {artifact.target} 的哈希无法对应需求的哈希，准备重新下载")
+            artifact.target.unlink()
+            await download_artifact(artifact)
+
+
+def register_artifacts(*artifacts: ArtifactDepends):
+    artifact_list.extend(artifacts)
+

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("utf-8"))
+        except ValidationError:
+            return self.cls()
+
+    def save(self, data: T):
+        self.fp.write_text(data.model_dump_json(), "utf-8")
+
+    @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/database/__init__.py

@@ -0,0 +1,231 @@
+from contextlib import asynccontextmanager
+import os
+import asyncio
+import sqlparse
+from pathlib import Path
+from typing import List, Dict, Any, Optional, Union, TYPE_CHECKING
+
+import aiosqlite
+
+if TYPE_CHECKING:
+    from . import DatabaseManager
+
+# 全局数据库管理器实例
+_global_db_manager: Optional["DatabaseManager"] = None
+
+
+def get_global_db_manager() -> "DatabaseManager":
+    """获取全局数据库管理器实例"""
+    global _global_db_manager
+    if _global_db_manager is None:
+        from . import DatabaseManager
+
+        _global_db_manager = DatabaseManager()
+    return _global_db_manager
+
+
+def close_global_db_manager() -> None:
+    """关闭全局数据库管理器实例"""
+    global _global_db_manager
+    if _global_db_manager is not None:
+        # 注意：这个函数应该在async环境中调用close_all_connections
+        _global_db_manager = None
+
+
+class DatabaseManager:
+    """异步数据库管理器"""
+
+    def __init__(self, db_path: Optional[Union[str, Path]] = None, pool_size: int = 5):
+        """
+        初始化数据库管理器
+
+        Args:
+            db_path: 数据库文件路径，支持str和Path类型
+            pool_size: 连接池大小
+        """
+        if db_path is None:
+            self.db_path = os.environ.get("DATABASE_PATH", "./data/database.db")
+        else:
+            self.db_path = str(db_path) if isinstance(db_path, Path) else db_path
+
+        # 连接池
+        self._connection_pool = []
+        self._pool_size = pool_size
+        self._lock = asyncio.Lock()
+        self._in_use = set()  # 跟踪正在使用的连接
+
+    async def _get_connection(self) -> aiosqlite.Connection:
+        """从连接池获取连接"""
+        async with self._lock:
+            # 尝试从池中获取现有连接
+            while self._connection_pool:
+                conn = self._connection_pool.pop()
+                # 检查连接是否仍然有效
+                try:
+                    await conn.execute("SELECT 1")
+                    self._in_use.add(conn)
+                    return conn
+                except:
+                    # 连接已失效，关闭它
+                    try:
+                        await conn.close()
+                    except:
+                        pass
+
+            # 如果连接池为空，创建新连接
+            conn = await aiosqlite.connect(self.db_path)
+            await conn.execute("PRAGMA foreign_keys = ON")
+            self._in_use.add(conn)
+            return conn
+
+    async def _return_connection(self, conn: aiosqlite.Connection) -> None:
+        """将连接返回到连接池"""
+        async with self._lock:
+            self._in_use.discard(conn)
+            if len(self._connection_pool) < self._pool_size:
+                self._connection_pool.append(conn)
+            else:
+                # 池已满，直接关闭连接
+                try:
+                    await conn.close()
+                except:
+                    pass
+
+    @asynccontextmanager
+    async def get_conn(self):
+        conn = await self._get_connection()
+        yield conn
+        await self._return_connection(conn)
+
+    async def query(
+        self, query: str, params: Optional[tuple] = None
+    ) -> List[Dict[str, Any]]:
+        """执行查询语句并返回结果"""
+        conn = await self._get_connection()
+        try:
+            cursor = await conn.execute(query, params or ())
+            columns = [description[0] for description in cursor.description]
+            rows = await cursor.fetchall()
+            results = [dict(zip(columns, row)) for row in rows]
+            await cursor.close()
+            return results
+        except Exception as e:
+            # 记录错误但重新抛出，让调用者处理
+            raise Exception(f"数据库查询失败: {str(e)}") from e
+        finally:
+            await self._return_connection(conn)
+
+    async def query_by_sql_file(
+        self, file_path: Union[str, Path], params: Optional[tuple] = None
+    ) -> List[Dict[str, Any]]:
+        """从 SQL 文件中读取查询语句并执行"""
+        path = str(file_path) if isinstance(file_path, Path) else file_path
+        with open(path, "r", encoding="utf-8") as f:
+            query = f.read()
+        return await self.query(query, params)
+
+    async def execute(self, command: str, params: Optional[tuple] = None) -> None:
+        """执行非查询语句"""
+        conn = await self._get_connection()
+        try:
+            await conn.execute(command, params or ())
+            await conn.commit()
+        except Exception as e:
+            # 记录错误但重新抛出，让调用者处理
+            raise Exception(f"数据库执行失败: {str(e)}") from e
+        finally:
+            await self._return_connection(conn)
+
+    async def execute_script(self, script: str) -> None:
+        """执行SQL脚本"""
+        conn = await self._get_connection()
+        try:
+            await conn.executescript(script)
+            await conn.commit()
+        except Exception as e:
+            # 记录错误但重新抛出，让调用者处理
+            raise Exception(f"数据库脚本执行失败: {str(e)}") from e
+        finally:
+            await self._return_connection(conn)
+
+    def _parse_sql_statements(self, script: str) -> List[str]:
+        """解析SQL脚本，分割成独立的语句"""
+        # 使用sqlparse库更准确地分割SQL语句
+        parsed = sqlparse.split(script)
+        statements = []
+
+        for statement in parsed:
+            statement = statement.strip()
+            if statement:
+                statements.append(statement)
+
+        return statements
+
+    async def execute_by_sql_file(
+        self,
+        file_path: Union[str, Path],
+        params: Optional[Union[tuple, List[tuple]]] = None,
+    ) -> None:
+        """从 SQL 文件中读取非查询语句并执行"""
+        path = str(file_path) if isinstance(file_path, Path) else file_path
+        with open(path, "r", encoding="utf-8") as f:
+            script = f.read()
+
+        # 如果有参数且是元组，使用execute执行整个脚本
+        if params is not None and isinstance(params, tuple):
+            await self.execute(script, params)
+        # 如果有参数且是列表，分别执行每个语句
+        elif params is not None and isinstance(params, list):
+            # 使用sqlparse准确分割SQL语句
+            statements = self._parse_sql_statements(script)
+            if len(statements) != len(params):
+                raise ValueError(
+                    f"语句数量({len(statements)})与参数组数量({len(params)})不匹配"
+                )
+
+            for statement, stmt_params in zip(statements, params):
+                if statement:
+                    await self.execute(statement, stmt_params)
+        # 如果无参数，使用executescript
+        else:
+            await self.execute_script(script)
+
+    async def execute_many(self, command: str, seq_of_params: List[tuple]) -> None:
+        """执行多条非查询语句"""
+        conn = await self._get_connection()
+        try:
+            await conn.executemany(command, seq_of_params)
+            await conn.commit()
+        except Exception as e:
+            # 记录错误但重新抛出，让调用者处理
+            raise Exception(f"数据库批量执行失败: {str(e)}") from e
+        finally:
+            await self._return_connection(conn)
+
+    async def execute_many_values_by_sql_file(
+        self, file_path: Union[str, Path], seq_of_params: List[tuple]
+    ) -> None:
+        """从 SQL 文件中读取一条语句，但是被不同值同时执行"""
+        path = str(file_path) if isinstance(file_path, Path) else file_path
+        with open(path, "r", encoding="utf-8") as f:
+            command = f.read()
+        await self.execute_many(command, seq_of_params)
+
+    async def close_all_connections(self) -> None:
+        """关闭所有连接"""
+        async with self._lock:
+            # 关闭池中的连接
+            for conn in self._connection_pool:
+                try:
+                    await conn.close()
+                except:
+                    pass
+            self._connection_pool.clear()
+
+            # 关闭正在使用的连接
+            for conn in self._in_use.copy():
+                try:
+                    await conn.close()
+                except:
+                    pass
+            self._in_use.clear()

konabot/common/llm/__init__.py

@@ -0,0 +1,67 @@
+from typing import Any, cast
+import openai
+
+from loguru import logger
+from openai.types.chat import ChatCompletion, ChatCompletionMessage, ChatCompletionMessageParam
+from pydantic import BaseModel, Field
+
+from konabot.common.path import CONFIG_PATH
+
+LLM_CONFIG_PATH = CONFIG_PATH / 'llm.json'
+
+if not LLM_CONFIG_PATH.exists():
+    LLM_CONFIG_PATH.write_text("{}")
+
+
+class LLMInfo(BaseModel):
+    base_url: str
+    api_key: str
+    model_name: str
+
+    def get_openai_client(self):
+        return openai.AsyncClient(
+            api_key=self.api_key,
+            base_url=self.base_url,
+        )
+
+    async def chat(
+        self,
+        messages: list[ChatCompletionMessageParam] | list[dict[str, Any]],
+        timeout: float | None = 30.0,
+        max_tokens: int | None = None,
+        **kwargs: Any,
+    ) -> ChatCompletionMessage:
+        logger.info(f"调用 LLM: BASE_URL={self.base_url} MODEL_NAME={self.model_name}")
+        completion: ChatCompletion = await self.get_openai_client().chat.completions.create(
+            messages=cast(Any, messages),
+            model=self.model_name,
+            max_tokens=max_tokens,
+            timeout=timeout,
+            stream=False,
+            **kwargs,
+        )
+        choice = completion.choices[0]
+        logger.info(
+            f"调用 LLM 完成: BASE_URL={self.base_url} MODEL_NAME={self.model_name} REASON={choice.finish_reason}"
+        )
+        return choice.message
+
+
+class LLMConfig(BaseModel):
+    llms: dict[str, LLMInfo] = Field(default_factory=dict)
+    default_llm: str = "Qwen2.5-7B-Instruct"
+
+
+llm_config = LLMConfig.model_validate_json(LLM_CONFIG_PATH.read_text())
+
+
+def get_llm(llm_model: str | None = None):
+    if llm_model is None:
+        llm_model = llm_config.default_llm
+    if llm_model not in llm_config.llms:
+        if llm_config.default_llm in llm_config.llms:
+            logger.warning(f"[LLM] 需求的 LLM 不存在，回退到默认模型 REQUIRED={llm_model}")
+            return llm_config.llms[llm_config.default_llm]
+        raise NotImplementedError("[LLM] LLM 未配置，该功能无法使用")
+    return llm_config.llms[llm_model]
+

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("utf-8")
+        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(), "utf-8")
+
+
+@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,227 @@
+from io import BytesIO
+from pathlib import Path
+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.discord import MessageEvent as DiscordMessageEvent
+from nonebot.adapters.discord.config import Config as DiscordConfig
+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 pydantic import BaseModel
+from returns.result import Failure, Result, Success
+
+
+discordConfig = nonebot.get_plugin_config(DiscordConfig)
+
+
+class ExtractImageConfig(BaseModel):
+    module_extract_image_no_download: bool = False
+    """
+    要不要算了，不下载了，直接爆炸算了，
+    适用于一些比较奇怪的网络环境，无法从协议端下载文件
+    """
+
+    module_extract_image_target: str = './assets/img/other/boom.jpg'
+    """
+    使用哪个图片呢
+    """
+
+
+module_config = nonebot.get_plugin_config(ExtractImageConfig)
+
+
+async def download_image_bytes(url: str, proxy: str | None = None) -> Result[bytes, str]:
+    # if "/matcha/cache/" in url:
+    #     url = url.replace('127.0.0.1', '10.126.126.101')
+    if module_config.module_extract_image_no_download:
+        return Success(Path(module_config.module_extract_image_target).read_bytes())
+    logger.debug(f"开始从 {url} 下载图片")
+    async with httpx.AsyncClient(proxy=proxy) 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_bytes(image: Image) -> Result[bytes, str]:
+    if image.url is not None:
+        raw_result = await download_image_bytes(image.url)
+    elif image.raw is not None:
+        if isinstance(image.raw, bytes):
+            raw_result = Success(image.raw)
+        else:
+            raw_result = Success(image.raw.getvalue())
+    else:
+        return Failure("由于一些内部问题，下载图片失败了orz")
+
+    return raw_result
+
+
+async def unimsg_img_to_pil(image: Image) -> Result[PIL.Image.Image, str]:
+    return (await unimsg_img_to_bytes(image)).bind(bytes_to_pil)
+
+
+async def extract_image_from_qq_message(
+    msg: OnebotV11Message,
+    evt: OnebotV11MessageEvent,
+    bot: OnebotV11Bot,
+    allow_reply: bool = True,
+) -> Result[bytes, 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("无法下载图片，可能有一些网络问题")
+            return await download_image_bytes(url)
+
+    return Failure("请在消息中包含图片，或者引用一个含有图片的消息")
+
+
+async def extract_image_data_from_message(
+    msg: Message,
+    evt: Event,
+    bot: Bot,
+    allow_reply: bool = True,
+) -> Result[bytes, 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)
+
+    if isinstance(evt, DiscordMessageEvent):
+        logger.debug('获取图片的路径方式走 Discord')
+        for a in evt.attachments:
+            if "image/" not in a.content_type:
+                continue
+            url = a.proxy_url
+            return await download_image_bytes(url, discordConfig.discord_proxy)
+
+    for seg in UniMessage.of(msg, bot):
+        logger.info(seg)
+        if isinstance(seg, Image):
+            return await unimsg_img_to_bytes(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_data_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_data(
+    evt: Event,
+    bot: Bot,
+    matcher: Matcher,
+) -> bytes | None:
+    match await extract_image_data_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
+
+
+async def _ext_img(
+    evt: Event,
+    bot: Bot,
+    matcher: Matcher,
+) -> PIL.Image.Image | None:
+    r = await _ext_img_data(evt, bot, matcher)
+    if r:
+        match bytes_to_pil(r):
+            case Success(img):
+                return img
+            case Failure(msg):
+                await matcher.send(await UniMessage.text(msg).export())
+    return None
+
+async def _try_ext_img(
+    evt: Event,
+    bot: Bot,
+    matcher: Matcher,
+) -> bytes | None:
+    match await extract_image_data_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
+
+DepImageBytes = Annotated[bytes, nonebot.params.Depends(_ext_img_data)]
+DepPILImage = Annotated[PIL.Image.Image, nonebot.params.Depends(_ext_img)]
+
+DepImageBytesOrNone = Annotated[bytes | None, nonebot.params.Depends(_try_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,33 @@
+from typing import Any, cast
+
+import nonebot
+from nonebot.adapters.onebot.v11 import Bot as OBBot
+from nonebot_plugin_alconna import UniMessage
+
+
+async def qq_broadcast(groups: list[str], msg: UniMessage[Any] | str):
+    if isinstance(msg, str):
+        msg = UniMessage.text(msg)
+    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/pager.py

@@ -0,0 +1,76 @@
+from dataclasses import dataclass
+from math import ceil
+from typing import Any, Callable
+
+from nonebot_plugin_alconna import UniMessage
+
+
+@dataclass
+class PagerQuery:
+    page_index: int
+    page_size: int
+
+    def apply[T](self, ls: list[T]) -> "PagerResult[T]":
+        if self.page_size <= 0:
+            return PagerResult(
+                success=False,
+                message="每页元素数量应该大于 0",
+                data=[],
+                page_count=-1,
+                query=self,
+            )
+        page_count = ceil(len(ls) / self.page_size)
+        if self.page_index <= 0 or self.page_size <= 0:
+            return PagerResult(
+                success=False,
+                message="页数必须大于 0",
+                data=[],
+                page_count=page_count,
+                query=self,
+            )
+        data = ls[(self.page_index - 1) * self.page_size: self.page_index * self.page_size]
+        if len(data) > 0:
+            return PagerResult(
+                success=True,
+                message="",
+                data=data,
+                page_count=page_count,
+                query=self,
+            )
+        return PagerResult(
+            success=False,
+            message="指定的页数超过最大页数",
+            data=data,
+            page_count=page_count,
+            query=self,
+        )
+
+
+@dataclass
+class PagerResult[T]:
+    data: list[T]
+    success: bool
+    message: str
+    page_count: int
+    query: PagerQuery
+
+    def to_unimessage(
+        self,
+        formatter: Callable[[T], str | UniMessage[Any]] = str,
+        title: str = '查询结果',
+        list_indicator: str = '- ',
+    ) -> UniMessage[Any]:
+        msg = UniMessage.text(f'===== {title} =====\n\n')
+
+        if not self.success:
+            msg = msg.text(f'⚠️ {self.message}\n')
+        else:
+            for obj in self.data:
+                msg = msg.text(list_indicator)
+                msg += formatter(obj)
+                msg += '\n'
+
+        msg = msg.text(f'\n===== 第 {self.query.page_index} 页，共 {self.page_count} 页 =====')
+        return msg
+
+

konabot/common/path.py

@@ -0,0 +1,28 @@
+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"
+TMP_PATH = DATA_PATH / "tmp"
+LOG_PATH = DATA_PATH / "logs"
+CONFIG_PATH = DATA_PATH / "config"
+BINARY_PATH = DATA_PATH / "bin"
+
+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()
+
+CONFIG_PATH.mkdir(exist_ok=True)
+TMP_PATH.mkdir(exist_ok=True)
+BINARY_PATH.mkdir(exist_ok=True)
+

konabot/common/permsys/__init__.py

@@ -0,0 +1,119 @@
+from typing import Annotated
+import nonebot
+from nonebot.adapters import Event
+from nonebot.params import Depends
+from nonebot.rule import Rule
+
+from konabot.common.appcontext import after_init
+from konabot.common.database import DatabaseManager
+from konabot.common.pager import PagerQuery
+from konabot.common.path import DATA_PATH
+from konabot.common.permsys.entity import PermEntity, get_entity_chain
+from konabot.common.permsys.migrates import execute_migration
+from konabot.common.permsys.repo import PermRepo
+
+
+db = DatabaseManager(DATA_PATH / "perm.sqlite3")
+_default_allow_permissions: set[str] = set()
+
+
+_EntityLike = Event | PermEntity | list[PermEntity]
+
+
+async def _to_entity_chain(el: _EntityLike):
+    if isinstance(el, Event):
+        return await get_entity_chain(el)  # pragma: no cover
+    if isinstance(el, PermEntity):
+        return [el]
+    return el
+
+
+class PermManager:
+    def __init__(self, db: DatabaseManager) -> None:
+        self.db = db
+
+    async def check_has_permission_info(self, entities: _EntityLike, key: str):
+        entities = await _to_entity_chain(entities)
+        key = key.removesuffix("*").removesuffix(".")
+        key_split = key.split(".")
+        key_split = [s for s in key_split if len(s) > 0]
+        keys = [".".join(key_split[: i + 1]) for i in range(len(key_split))][::-1] + [
+            "*"
+        ]
+
+        async with self.db.get_conn() as conn:
+            repo = PermRepo(conn)
+            data = await repo.get_perm_info_batch(entities, keys)
+            for entity in entities:
+                for k in keys:
+                    p = data.get((entity, k))
+                    if p is not None:
+                        return (entity, k, p)
+        return None
+
+    async def check_has_permission(self, entities: _EntityLike, key: str) -> bool:
+        res = await self.check_has_permission_info(entities, key)
+        if res is None:
+            return False
+        return res[2]
+
+    async def update_permission(self, entity: PermEntity, key: str, perm: bool | None):
+        async with self.db.get_conn() as conn:
+            repo = PermRepo(conn)
+            await repo.update_perm_info(entity, key, perm)
+
+    async def list_permission(self, entities: _EntityLike, query: PagerQuery):
+        entities = await _to_entity_chain(entities)
+        async with self.db.get_conn() as conn:
+            repo = PermRepo(conn)
+            return await repo.list_perm_info_batch(entities, query)
+
+
+def perm_manager(_db: DatabaseManager | None = None) -> PermManager:  # pragma: no cover
+    if _db is None:
+        _db = db
+    return PermManager(_db)
+
+
+@after_init
+def create_startup():  # pragma: no cover
+    from konabot.common.nb.is_admin import cfg
+
+    driver = nonebot.get_driver()
+
+    @driver.on_startup
+    async def _():
+        async with db.get_conn() as conn:
+            await execute_migration(conn)
+            pm = perm_manager(db)
+            for account in cfg.admin_qq_account:
+                # ^ 这里的是超级管理员！！用环境变量定义的。
+                #   咕嘿嘿嘿！！！夺取全部权限！！！
+                await pm.update_permission(
+                    PermEntity("ob11", "user", str(account)), "*", True
+                )
+            for key in _default_allow_permissions:
+                await pm.update_permission(
+                    PermEntity("sys", "global", "global"), key, True
+                )
+
+    @driver.on_shutdown
+    async def _():
+        try:
+            await db.close_all_connections()
+        except Exception:
+            pass
+
+
+DepPermManager = Annotated[PermManager, Depends(perm_manager)]
+
+
+def register_default_allow_permission(key: str):
+    _default_allow_permissions.add(key)
+
+
+def require_permission(perm: str) -> Rule:  # pragma: no cover
+    async def check_permission(event: Event, pm: DepPermManager) -> bool:
+        return await pm.check_has_permission(event, perm)
+
+    return Rule(check_permission)

konabot/common/permsys/entity.py

@@ -0,0 +1,69 @@
+from dataclasses import dataclass
+from nonebot.internal.adapter import Event
+
+from nonebot.adapters.onebot.v11 import Event as OB11Event
+from nonebot.adapters.onebot.v11.event import GroupMessageEvent as OB11GroupEvent
+from nonebot.adapters.onebot.v11.event import PrivateMessageEvent as OB11PrivateEvent
+
+from nonebot.adapters.discord.event import Event as DiscordEvent
+from nonebot.adapters.discord.event import GuildMessageCreateEvent as DiscordGMEvent
+from nonebot.adapters.discord.event import DirectMessageCreateEvent as DiscordDMEvent
+
+from nonebot.adapters.minecraft.event import MessageEvent as MinecraftMessageEvent
+
+from nonebot.adapters.console.event import MessageEvent as ConsoleEvent
+
+
+@dataclass(frozen=True)
+class PermEntity:
+    platform: str
+    entity_type: str
+    external_id: str
+
+
+def get_entity_chain_of_entity(entity: PermEntity) -> list[PermEntity]:
+    return [
+        PermEntity("sys", "global", "global"),
+        PermEntity(entity.platform, "global", "global"),
+        entity,
+    ][::-1]
+
+
+async def get_entity_chain(event: Event) -> list[PermEntity]:  # pragma: no cover
+    entities = [PermEntity("sys", "global", "global")]
+
+    if isinstance(event, OB11Event):
+        entities.append(PermEntity("ob11", "global", "global"))
+
+    if isinstance(event, OB11GroupEvent):
+        entities.append(PermEntity("ob11", "group", str(event.group_id)))
+        entities.append(PermEntity("ob11", "user", str(event.user_id)))
+
+    if isinstance(event, OB11PrivateEvent):
+        entities.append(PermEntity("ob11", "user", str(event.user_id)))
+
+    if isinstance(event, DiscordEvent):
+        entities.append(PermEntity("discord", "global", "global"))
+
+    if isinstance(event, DiscordGMEvent):
+        entities.append(PermEntity("discord", "guild", str(event.guild_id)))
+        entities.append(PermEntity("discord", "channel", str(event.channel_id)))
+        entities.append(PermEntity("discord", "user", str(event.user_id)))
+
+    if isinstance(event, DiscordDMEvent):
+        entities.append(PermEntity("discord", "channel", str(event.channel_id)))
+        entities.append(PermEntity("discord", "user", str(event.user_id)))
+
+    if isinstance(event, MinecraftMessageEvent):
+        entities.append(PermEntity("minecraft", "global", "global"))
+        entities.append(PermEntity("minecraft", "server", event.server_name))
+        player_uuid = event.player.uuid
+        if player_uuid is not None:
+            entities.append(PermEntity("minecraft", "player", player_uuid.hex))
+
+    if isinstance(event, ConsoleEvent):
+        entities.append(PermEntity("console", "global", "global"))
+        entities.append(PermEntity("console", "channel", event.channel.id))
+        entities.append(PermEntity("console", "user", event.user.id))
+
+    return entities[::-1]

konabot/common/permsys/migrates/__init__.py

@@ -0,0 +1,81 @@
+from dataclasses import dataclass
+from pathlib import Path
+
+import aiosqlite
+from loguru import logger
+
+from konabot.common.database import DatabaseManager
+from konabot.common.path import DATA_PATH
+
+
+PATH_THISFOLDER = Path(__file__).parent
+
+SQL_CHECK_EXISTS = (PATH_THISFOLDER / "./check_migrate_version_exists.sql").read_text()
+SQL_CREATE_TABLE = (PATH_THISFOLDER / "./create_migrate_version_table.sql").read_text()
+SQL_GET_MIGRATE_VERSION = (PATH_THISFOLDER / "get_migrate_version.sql").read_text()
+SQL_UPDATE_VERSION = (PATH_THISFOLDER / "./update_migrate_version.sql").read_text()
+
+db = DatabaseManager(DATA_PATH / "perm.sqlite3")
+
+
+@dataclass
+class Migration:
+    upgrade: str | Path
+    downgrade: str | Path
+
+    def get_upgrade_script(self) -> str:
+        if isinstance(self.upgrade, Path):
+            return self.upgrade.read_text()
+        return self.upgrade
+
+    def get_downgrade_script(self) -> str:
+        if isinstance(self.downgrade, Path):
+            return self.downgrade.read_text()
+        return self.downgrade
+
+
+migrations = [
+    Migration(
+        PATH_THISFOLDER / "./mu1_create_permsys_table.sql",
+        PATH_THISFOLDER / "./md1_remove_permsys_table.sql",
+    ),
+]
+
+
+TARGET_VERSION = len(migrations)
+
+
+async def get_current_version(conn: aiosqlite.Connection) -> int:
+    cursor = await conn.execute(SQL_CHECK_EXISTS)
+    count = await cursor.fetchone()
+    assert count is not None
+    if count[0] < 1:
+        logger.info("权限系统数据表不存在，现在创建表")
+        await conn.executescript(SQL_CREATE_TABLE)
+        await conn.commit()
+        return 0
+    cursor = await conn.execute(SQL_GET_MIGRATE_VERSION)
+    row = await cursor.fetchone()
+    if row is None:
+        return 0
+    return row[0]
+
+
+async def execute_migration(
+    conn: aiosqlite.Connection,
+    version: int = TARGET_VERSION,
+    migrations: list[Migration] = migrations,
+):
+    now_version = await get_current_version(conn)
+    while now_version < version:
+        migration = migrations[now_version]
+        await conn.executescript(migration.get_upgrade_script())
+        now_version += 1
+        await conn.execute(SQL_UPDATE_VERSION, (now_version,))
+        await conn.commit()
+    while now_version > version:
+        migration = migrations[now_version - 1]
+        await conn.executescript(migration.get_downgrade_script())
+        now_version -= 1
+        await conn.execute(SQL_UPDATE_VERSION, (now_version,))
+        await conn.commit()

konabot/common/permsys/migrates/check_migrate_version_exists.sql

@@ -0,0 +1,7 @@
+SELECT
+    COUNT(*)
+FROM
+    sqlite_master
+WHERE
+    type = 'table'
+    AND name = 'migrate_version'

konabot/common/permsys/migrates/create_migrate_version_table.sql

@@ -0,0 +1,3 @@
+CREATE TABLE migrate_version(version INT PRIMARY KEY);
+INSERT INTO migrate_version(version)
+VALUES(0);

konabot/common/permsys/migrates/get_migrate_version.sql

@@ -0,0 +1,4 @@
+SELECT
+    version
+FROM
+    migrate_version;

konabot/common/permsys/migrates/md1_remove_permsys_table.sql

@@ -0,0 +1,2 @@
+DROP TABLE IF EXISTS perm_entity;
+DROP TABLE IF EXISTS perm_info;

konabot/common/permsys/migrates/mu1_create_permsys_table.sql

@@ -0,0 +1,30 @@
+CREATE TABLE perm_entity(
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    platform TEXT NOT NULL,
+    entity_type TEXT NOT NULL,
+    external_id TEXT NOT NULL,
+    created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
+    updated_at DATETIME DEFAULT CURRENT_TIMESTAMP
+);
+
+CREATE UNIQUE INDEX idx_perm_entity_lookup
+ON perm_entity(platform, entity_type, external_id);
+
+CREATE TABLE perm_info(
+    entity_id INTEGER NOT NULL,
+    config_key TEXT NOT NULL,
+    value BOOLEAN,
+    updated_at DATETIME DEFAULT CURRENT_TIMESTAMP,
+    -- 联合主键
+    PRIMARY KEY (entity_id, config_key)
+);
+
+CREATE TRIGGER perm_entity_update AFTER UPDATE
+ON perm_entity BEGIN
+    UPDATE perm_entity SET updated_at=CURRENT_TIMESTAMP WHERE id=old.id;
+END;
+CREATE TRIGGER perm_info_update AFTER UPDATE
+ON perm_info BEGIN
+    UPDATE perm_info SET updated_at=CURRENT_TIMESTAMP WHERE entity_id=old.entity_id;
+END;
+

konabot/common/permsys/migrates/update_migrate_version.sql

@@ -0,0 +1,2 @@
+UPDATE migrate_version
+SET version = ?;

konabot/common/permsys/repo.py

@@ -0,0 +1,242 @@
+from dataclasses import dataclass
+import math
+from pathlib import Path
+
+import aiosqlite
+
+from konabot.common.pager import PagerQuery, PagerResult
+
+from .entity import PermEntity
+
+
+def s(p: str):
+    """读取 SQL 文件内容。
+
+    Args:
+        p: SQL 文件名（相对于当前文件所在目录的 sql/ 子目录）。
+
+    Returns:
+        SQL 文件的内容字符串。
+    """
+    return (Path(__file__).parent / "./sql/" / p).read_text()
+
+
+@dataclass
+class PermRepo:
+    """权限实体存储库，负责与数据库交互管理权限实体。
+
+    Attributes:
+        conn: aiosqlite 数据库连接对象。
+    """
+
+    conn: aiosqlite.Connection
+
+    async def create_entity(self, entity: PermEntity) -> int:
+        """创建新的权限实体并返回其 ID。
+
+        Args:
+            entity: 要创建的权限实体对象。
+
+        Returns:
+            新创建实体的数据库 ID。
+
+        Raises:
+            AssertionError: 如果创建后无法获取实体 ID。
+        """
+        await self.conn.execute(
+            s("create_entity.sql"),
+            (entity.platform, entity.entity_type, entity.external_id),
+        )
+        await self.conn.commit()
+        eid = await self._get_entity_id_or_none(entity)
+        assert eid is not None
+        return eid
+
+    async def _get_entity_id_or_none(self, entity: PermEntity) -> int | None:
+        """查询实体 ID，如果不存在则返回 None。
+
+        Args:
+            entity: 要查询的权限实体对象。
+
+        Returns:
+            实体 ID，如果不存在则返回 None。
+        """
+        res = await self.conn.execute(
+            s("get_entity_id.sql"),
+            (entity.platform, entity.entity_type, entity.external_id),
+        )
+        row = await res.fetchone()
+        if row is None:
+            return None
+        return row[0]
+
+    async def get_entity_id(self, entity: PermEntity) -> int:
+        """获取实体 ID，如果不存在则自动创建。
+
+        Args:
+            entity: 权限实体对象。
+
+        Returns:
+            实体的数据库 ID。
+        """
+        eid = await self._get_entity_id_or_none(entity)
+        if eid is None:
+            return await self.create_entity(entity)
+        return eid
+
+    async def get_perm_info(self, entity: PermEntity, config_key: str) -> bool | None:
+        """获取实体的权限配置信息。
+
+        Args:
+            entity: 权限实体对象。
+            config_key: 配置项的键名。
+
+        Returns:
+            配置值（True/False），如果不存在则返回 None。
+        """
+        eid = await self.get_entity_id(entity)
+        res = await self.conn.execute(
+            s("get_perm_info.sql"),
+            (eid, config_key),
+        )
+        row = await res.fetchone()
+        if row is None:
+            return None
+        return bool(row[0])
+
+    async def update_perm_info(
+        self, entity: PermEntity, config_key: str, value: bool | None
+    ):
+        """更新实体的权限配置信息。
+
+        Args:
+            entity: 权限实体对象。
+            config_key: 配置项的键名。
+            value: 要设置的配置值（True/False/None）。
+        """
+        eid = await self.get_entity_id(entity)
+        await self.conn.execute(s("update_perm_info.sql"), (eid, config_key, value))
+        await self.conn.commit()
+
+    async def get_entity_id_batch(
+        self, entities: list[PermEntity]
+    ) -> dict[PermEntity, int]:
+        """批量获取 Entity 的 entity_id
+
+        Args:
+            entities: PermEntity 列表
+
+        Returns:
+            字典，键为 PermEntity，值为对应的 ID
+        """
+
+        # for entity in entities:
+        #     await self.conn.execute(
+        #         s("create_entity.sql"),
+        #         (entity.platform, entity.entity_type, entity.external_id),
+        #     )
+        await self.conn.executemany(
+            s("create_entity.sql"),
+            [(e.platform, e.entity_type, e.external_id) for e in entities],
+        )
+        await self.conn.commit()
+        val_placeholders = ", ".join(["(?, ?, ?)"] * len(entities))
+        params = []
+        for e in entities:
+            params.extend([e.platform, e.entity_type, e.external_id])
+        cursor = await self.conn.execute(
+            f"""
+                SELECT id, platform, entity_type, external_id
+                FROM perm_entity
+                WHERE (platform, entity_type, external_id) IN (VALUES {val_placeholders});
+            """,
+            params,
+        )
+        rows = await cursor.fetchall()
+        return {PermEntity(row[1], row[2], row[3]): row[0] for row in rows}
+
+    async def get_perm_info_batch(
+        self, entities: list[PermEntity], config_keys: list[str]
+    ) -> dict[tuple[PermEntity, str], bool]:
+        """批量获取权限信息
+
+        Args:
+            entities: PermEntity 列表
+            config_keys: 查询的键列表
+
+        Returns:
+            字典，键是 PermEntity 和 config_key 的元组，值是布尔，过滤掉所有空值
+        """
+        entity_ids = {
+            v: k for k, v in (await self.get_entity_id_batch(entities)).items()
+        }
+        placeholders1 = ", ".join("?" * len(entity_ids))
+        placeholders2 = ", ".join("?" * len(config_keys))
+        sql = f"""
+            SELECT entity_id, config_key, value
+            FROM perm_info
+            WHERE entity_id IN ({placeholders1})
+              AND config_key IN ({placeholders2})
+              AND value IS NOT NULL;
+        """
+
+        params = tuple(entity_ids.keys()) + tuple(config_keys)
+        cursor = await self.conn.execute(sql, params)
+        rows = await cursor.fetchall()
+
+        return {(entity_ids[row[0]], row[1]): bool(row[2]) for row in rows}
+
+    async def list_perm_info_batch(
+        self, entities: list[PermEntity], pager: PagerQuery
+    ) -> PagerResult[tuple[PermEntity, str, bool]]:
+        """批量获取某个实体的权限信息
+
+        Args:
+            entities: PermEntity 列表
+            pager: PagerQuery 对象，即分页要求
+
+        Returns:
+            字典，键是 PermEntity，值是权限条目和布尔的元组，过滤掉所有空值
+        """
+        entity_to_id = await self.get_entity_id_batch(entities)
+        id_to_entity = {v: k for k, v in entity_to_id.items()}
+        ordered_ids = [entity_to_id[e] for e in entities if e in entity_to_id]
+
+        placeholders = ", ".join("?" * len(ordered_ids))
+        order_by_cases = " ".join([f"WHEN ? THEN {i}" for i in range(len(ordered_ids))])
+
+        pagecount_sql = f"SELECT COUNT(*) FROM perm_info WHERE entity_id IN ({placeholders}) AND value IS NOT NULL;"
+        count_cursor = await self.conn.execute(pagecount_sql, tuple(ordered_ids))
+        total_count = (await count_cursor.fetchone() or (0,))[0]
+
+        sql = f"""
+            SELECT entity_id, config_key, value
+            FROM perm_info
+            WHERE entity_id IN ({placeholders})
+              AND value IS NOT NULL
+            ORDER BY
+                (CASE entity_id {order_by_cases} END) ASC,
+                config_key ASC
+            LIMIT ?
+            OFFSET ?;
+        """
+
+        params = (
+            tuple(ordered_ids)
+            + tuple(ordered_ids)
+            + (
+                pager.page_size,
+                (pager.page_index - 1) * pager.page_size,
+            )
+        )
+        cursor = await self.conn.execute(sql, params)
+        rows = await cursor.fetchall()
+
+        # return {entity_ids[row[0]]: (row[1], bool(row[2])) for row in rows}
+        return PagerResult(
+            data=[(id_to_entity[row[0]], row[1], row[2]) for row in rows],
+            success=True,
+            message="",
+            page_count=math.ceil(total_count / pager.page_size),
+            query=pager,
+        )

konabot/common/permsys/sql/create_entity.sql

@@ -0,0 +1,11 @@
+INSERT
+OR IGNORE INTO perm_entity(
+    platform,
+    entity_type,
+    external_id
+)
+VALUES(
+    ?,
+    ?,
+    ?
+);