diff --git a/PLUGINS.md b/PLUGINS.md index 848117e..e7d4247 100644 --- a/PLUGINS.md +++ b/PLUGINS.md @@ -1,172 +1,173 @@ -# 蛋定助手插件文档 - -## 项目概述 - -蛋定助手是一个基于 NoneBot2 框架开发的 QQ 机器人，提供多种功能插件，包括 AI 聊天、游戏辅助、积分系统、社群管理等。 - -## 插件总览 - -| 插件名称 | 描述 | 触发方式 | 权限要求 | -|---------|------|---------|---------| -| [chatai](#1-chatai---ai-聊天) | AI 聊天（DeepSeek），支持图文回复 | `*` 开头消息 | 所有用户 | -| [auto_recall](#2-auto_recall---自动撤回) | 自动撤回机器人发送的消息 | 自动执行 | 系统自动 | -| [auto_friend_accept](#3-auto_friend_accept---自动接受好友) | 自动同意好友请求并发送欢迎语 | 自动执行 | 系统自动 | -| [welcome_plugin](#4-welcome_plugin---入群欢迎) | 新成员入群欢迎并发送帮助菜单 | 自动执行 | 特定群 (621016172) | -| [danding_qqpush](#5-danding_qqpush---消息推送) | 通过 HTTP API 向指定群推送图文通知 | HTTP POST | 接口 Token 验证 | -| [danding_points](#6-danding_points---积分系统核心) | 积分系统数据库与 API 核心 | API 调用 | 系统内部 | -| [danding_points_query](#7-danding_points_query---积分查询) | 查询余额、排行及交易记录 | 命令触发 | 所有用户 | -| [group_horse_racing](#8-group_horse_racing---群赛马游戏) | 多人赛马游戏，支持下注与积分奖惩 | 命令触发 | 允许的群聊 | -| [onmyoji_gacha](#9-onmyoji_gacha---阴阳师抽卡模拟) | 阴阳师主题抽卡，包含成就与卡密奖励 | 命令触发 | 允许的群聊/用户 | -| [damo_balance](#10-damo_balance---大漠账户查询) | 查询大漠平台账户余额 | 命令触发 | 特定用户 | -| [danding_api](#11-danding_api---管理-api) | 管理员操作：卡密管理、加时、在线查询 | 命令触发 | 超级用户 | -| [danding_help](#12-danding_help---帮助菜单) | 提供教程、下载及功能指引 | 命令触发 | 特定群 (621016172) | -| [command_list](#13-command_list---指令列表) | 获取系统支持的所有指令列表 | 命令触发 | 所有用户 | -| [danding_sign_in](#14-danding_sign_in---每日签到) | 独立每日签到，与抽卡签到共用数据 | 命令触发 | 白名单群聊/管理员 | - ---- - -## 详细插件文档 - -### 1. chatai - AI 聊天 - -基于 DeepSeek AI 的聊天功能，支持将回复转换为图片形式，并在一定时间后自动撤回。 - -- **使用方法**: 发送以 `*` 开头的消息。 -- **配置项**: `DEEPSEEK_TOKEN` (必填)。 -- **特性**: AI 回复会自动转为图片显示，默认 120 秒后撤回。 - -### 2. auto_recall - 自动撤回 - -监控机器人发出的消息，并在指定时间后自动撤回，保持聊天环境整洁。 - -- **配置项**: - - `RECALL_DELAY`: 普通消息撤回延迟（默认 110s）。 - - `QQPUSH_RECALL_DELAY`: 推送消息撤回延迟（默认 3600s）。 - -### 3. auto_friend_accept - 自动接受好友 - -自动处理好友请求，提升用户接入效率。 - -- **配置项**: - - `auto_accept_enabled`: 是否开启自动接受。 - - `auto_reply_message`: 接受后的欢迎语。 - -### 4. welcome_plugin - 入群欢迎 - -针对特定群聊的新成员欢迎功能。 - -- **触发场景**: 新成员加入群 `621016172`。 -- **功能**: 随机发送欢迎语，并附带帮助菜单图片。 - -### 5. danding_qqpush - 消息推送 - -提供外部系统向 QQ 推送通知的 HTTP 接口。 - -- **接口**: `POST /danding/qqpush/{token}` -- **功能**: 自动将长文本转换为图片，支持 `@用户` 和换行符 `#`。 -- **配置**: `DANDING_QQPUSH_TOKEN`。 - -### 6. danding_points - 积分系统核心 - -为其他插件提供积分存储与结算的基础设施。 - -- **功能**: 数据库管理（SQLite）、余额增减、排行榜计算、交易日志记录。 -- **数据库路径**: `data/danding_points/points.db` - -### 7. danding_points_query - 积分查询 - -用户通过命令与积分系统交互。 - -- **主要命令**: - - `我的积分`: 查看个人余额。 - - `积分查询 @用户`: 查看他人余额。 - - `积分排行`: 查看前 10 名。 - - `积分历史查询`: 查看最近 5 条变动记录。 - - `积分帮助`: 获取指令指引。 - -### 8. group_horse_racing - 群赛马游戏 - -集成积分系统的多人互动游戏。 - -- **主要命令**: - - `/赛马报名 [马名]`: 参加比赛。 - - `/赛马下注 <序号> <金额>`: 对马匹下注。 - - `/赛马开赛`: 开始比赛（至少 2 人）。 -- **积分逻辑**: 参赛奖励 50 分，冠军奖励 200 分，支持下注赔率结算。 - -### 9. onmyoji_gacha - 阴阳师抽卡模拟 - -高度还原的抽卡模拟，包含成就系统。采用模块化架构，职责分明。 - -- **模块结构**: - - `__init__.py`: 路由注册与matcher定义（167行） - - `config.py`: Pydantic配置管理 - - `gacha.py`: 抽卡核心逻辑（GachaSystem类） - - `data_manager.py`: SQLite数据持久化（DataManager类） - - `rules.py`: 命令匹配规则（check_permission等） - - `formatters.py`: 消息格式化（9个格式化函数） - - `handlers/`: 命令处理函数（9个handler模块） - - `utils.py`: 通用工具函数 - - `api_utils.py`: 积分系统API交互 - - `web_api.py`: HTTP API路由 - -- **主要命令**: - - `抽卡`: 执行单抽。 - - `三连抽`: 执行三连抽。 - - `我的抽卡`: 查看个人统计。 - - `抽卡排行`: 查看抽卡排行榜。 - - `今日抽卡`: 查看今日抽卡统计。 - - `查询成就`: 查看已解锁成就及进度（非酋、勤勤恳恳系列）。 - - `抽卡介绍`: 查看详细机制与奖励说明。 - - `积分商店`（`兑换商店`）: 查看可兑换商品、余额与剩余抽卡次数。 - - `兑换抽卡`（`积分兑换`）: 消耗 99 Bot 积分兑换 1 次当日抽卡机会。 -- **特性**: 抽中 SSR/SP 可获得"蛋定助手"卡密奖励（需联系管理员领取）。包含每日抽卡签到积分奖励（与独立「签到」命令、WPF 客户端共用同一份签到数据，每天仅一次）。成就系统自动发放积分奖励。 -### 10. damo_balance - 大漠账户查询 - -查询大漠平台账户余额。 - -- **命令**: `大漠余额` 或 `余额查询`。 -- **限制**: 仅特定用户可用，需输入验证码。 - -### 11. danding_api - 管理 API - -供超级用户使用的后台管理功能。 - -- **主要命令**: - - `在线人数`: 查询当前活跃用户。 - - `生成卡密 <类型>`: 生成天/周/月卡。 - - `用户加时 <用户名> <类型>`: 直接为特定用户增加时长。 - -### 12. danding_help - 帮助菜单 - -系统的官方指引手册。 - -- **主要命令**: `帮助`、`下载`、`活动`（别名：`最新活动`、`群活动`、`hd`）、`每日试用`、`正式版如何运行` 等。 -- **活动海报推送**: 发送 `活动` 时自动请求服务端接口获取当前正在进行中的运营活动，并向群内发送对应的活动宣传海报及说明。 -- **限制**: 仅在特定群 `621016172` 可用。 - -### 13. command_list - 指令列表 - -快速查阅所有可用指令。 - -- **命令**: `指令列表`。 - -### 14. danding_sign_in - 每日签到 - -独立的每日签到命令，与抽卡签到、WPF 客户端签到**共用同一份签到数据**（xapi `bot_gacha_daily_sign_in` 表，每用户每天仅一次，任一入口先到先得）。 - -- **主要命令**: `签到`（别名：`每日签到`、`打卡`）。 -- **积分奖励**: 随机 1-100 积分（与抽卡签到一致），积分流水 `source=gacha_sign`、`reason=抽卡签到`。 -- **权限**: 白名单群聊 `621016172` 或管理员可用；私聊非管理员不可用。 -- **数据共用说明**: 同一天内，抽卡/WPF/命令签到任一入口先触发者获得积分，后触发者提示"今天已签到"，不会重复发放。 - ---- - -## 常见问题 (FAQ) - -- **Q: 为什么某些命令没反应？** - A: 部分插件（如 `danding_help`）限制了特定群聊使用；管理指令需要配置 `SUPERUSERS`。 -- **Q: 积分有什么用？** - A: 目前主要用于赛马下注及展示排名。 -- **Q: 抽卡奖励如何领取？** - A: 抽中 SSR/SP 或解锁特定成就后，请截屏联系管理员。 +# 蛋定助手插件文档 + +## 项目概述 + +蛋定助手是一个基于 NoneBot2 框架开发的 QQ 机器人，提供多种功能插件，包括 AI 聊天、游戏辅助、积分系统、社群管理等。 + +## 插件总览 + +| 插件名称 | 描述 | 触发方式 | 权限要求 | +|---------|------|---------|---------| +| [chatai](#1-chatai---ai-聊天) | AI 聊天（DeepSeek），支持图文回复 | `*` 开头消息 | 所有用户 | +| [auto_recall](#2-auto_recall---自动撤回) | 自动撤回机器人发送的消息 | 自动执行 | 系统自动 | +| [auto_friend_accept](#3-auto_friend_accept---自动接受好友) | 自动同意好友请求并发送欢迎语 | 自动执行 | 系统自动 | +| [welcome_plugin](#4-welcome_plugin---入群欢迎) | 新成员入群欢迎并发送帮助菜单 | 自动执行 | 特定群 (621016172) | +| [danding_qqpush](#5-danding_qqpush---消息推送) | 通过 HTTP API 向指定群推送图文通知；带超级用户预览命令 | HTTP POST / 命令触发 | 接口 Token 验证 / 超级用户 | +| [danding_points](#6-danding_points---积分系统核心) | 积分系统数据库与 API 核心 | API 调用 | 系统内部 | +| [danding_points_query](#7-danding_points_query---积分查询) | 查询余额、排行及交易记录 | 命令触发 | 所有用户 | +| [group_horse_racing](#8-group_horse_racing---群赛马游戏) | 多人赛马游戏，支持下注与积分奖惩 | 命令触发 | 允许的群聊 | +| [onmyoji_gacha](#9-onmyoji_gacha---阴阳师抽卡模拟) | 阴阳师主题抽卡，包含成就与卡密奖励 | 命令触发 | 允许的群聊/用户 | +| [damo_balance](#10-damo_balance---大漠账户查询) | 查询大漠平台账户余额 | 命令触发 | 特定用户 | +| [danding_api](#11-danding_api---管理-api) | 管理员操作：卡密管理、加时、在线查询 | 命令触发 | 超级用户 | +| [danding_help](#12-danding_help---帮助菜单) | 提供教程、下载及功能指引 | 命令触发 | 特定群 (621016172) | +| [command_list](#13-command_list---指令列表) | 获取系统支持的所有指令列表 | 命令触发 | 所有用户 | +| [danding_sign_in](#14-danding_sign_in---每日签到) | 独立每日签到，与抽卡签到共用数据 | 命令触发 | 白名单群聊/管理员 | + +--- + +## 详细插件文档 + +### 1. chatai - AI 聊天 + +基于 DeepSeek AI 的聊天功能，支持将回复转换为图片形式，并在一定时间后自动撤回。 + +- **使用方法**: 发送以 `*` 开头的消息。 +- **配置项**: `DEEPSEEK_TOKEN` (必填)。 +- **特性**: AI 回复会自动转为图片显示，默认 120 秒后撤回。 + +### 2. auto_recall - 自动撤回 + +监控机器人发出的消息，并在指定时间后自动撤回，保持聊天环境整洁。 + +- **配置项**: + - `RECALL_DELAY`: 普通消息撤回延迟（默认 110s）。 + - `QQPUSH_RECALL_DELAY`: 推送消息撤回延迟（默认 3600s）。 + +### 3. auto_friend_accept - 自动接受好友 + +自动处理好友请求，提升用户接入效率。 + +- **配置项**: + - `auto_accept_enabled`: 是否开启自动接受。 + - `auto_reply_message`: 接受后的欢迎语。 + +### 4. welcome_plugin - 入群欢迎 + +针对特定群聊的新成员欢迎功能。 + +- **触发场景**: 新成员加入群 `621016172`。 +- **功能**: 随机发送欢迎语，并附带帮助菜单图片。 + +### 5. danding_qqpush - 消息推送 + +提供外部系统向 QQ 推送通知的 HTTP 接口。 + +- **接口**: `POST /danding/qqpush/{token}` +- **功能**: 自动将长文本转换为图片，支持 `@用户` 和换行符 `#`。 +- **配置**: `DANDING_QQPUSH_TOKEN`。 +- **测试命令**: `测试通知图片`（别名 `通知图片测试`、`测试通知卡片`），仅超级用户可用；群聊中渲染群聊通知卡片、私聊中渲染私聊通知卡片，用于预览固定示例文本（含标题、中文、多行、Emoji 与中文标点）的版式，不调用 HTTP 接口、不新增配置项。 + +### 6. danding_points - 积分系统核心 + +为其他插件提供积分存储与结算的基础设施。 + +- **功能**: 数据库管理（SQLite）、余额增减、排行榜计算、交易日志记录。 +- **数据库路径**: `data/danding_points/points.db` + +### 7. danding_points_query - 积分查询 + +用户通过命令与积分系统交互。 + +- **主要命令**: + - `我的积分`: 查看个人余额。 + - `积分查询 @用户`: 查看他人余额。 + - `积分排行`: 查看前 10 名。 + - `积分历史查询`: 查看最近 5 条变动记录。 + - `积分帮助`: 获取指令指引。 + +### 8. group_horse_racing - 群赛马游戏 + +集成积分系统的多人互动游戏。 + +- **主要命令**: + - `/赛马报名 [马名]`: 参加比赛。 + - `/赛马下注 <序号> <金额>`: 对马匹下注。 + - `/赛马开赛`: 开始比赛（至少 2 人）。 +- **积分逻辑**: 参赛奖励 50 分，冠军奖励 200 分，支持下注赔率结算。 + +### 9. onmyoji_gacha - 阴阳师抽卡模拟 + +高度还原的抽卡模拟，包含成就系统。采用模块化架构，职责分明。 + +- **模块结构**: + - `__init__.py`: 路由注册与matcher定义（167行） + - `config.py`: Pydantic配置管理 + - `gacha.py`: 抽卡核心逻辑（GachaSystem类） + - `data_manager.py`: SQLite数据持久化（DataManager类） + - `rules.py`: 命令匹配规则（check_permission等） + - `formatters.py`: 消息格式化（9个格式化函数） + - `handlers/`: 命令处理函数（9个handler模块） + - `utils.py`: 通用工具函数 + - `api_utils.py`: 积分系统API交互 + - `web_api.py`: HTTP API路由 + +- **主要命令**: + - `抽卡`: 执行单抽。 + - `三连抽`: 执行三连抽。 + - `我的抽卡`: 查看个人统计。 + - `抽卡排行`: 查看抽卡排行榜。 + - `今日抽卡`: 查看今日抽卡统计。 + - `查询成就`: 查看已解锁成就及进度（非酋、勤勤恳恳系列）。 + - `抽卡介绍`: 查看详细机制与奖励说明。 + - `积分商店`（`兑换商店`）: 查看可兑换商品、余额与剩余抽卡次数。 + - `兑换抽卡`（`积分兑换`）: 消耗 99 Bot 积分兑换 1 次当日抽卡机会。 +- **特性**: 抽中 SSR/SP 可获得"蛋定助手"卡密奖励（需联系管理员领取）。包含每日抽卡签到积分奖励（与独立「签到」命令、WPF 客户端共用同一份签到数据，每天仅一次）。成就系统自动发放积分奖励。 +### 10. damo_balance - 大漠账户查询 + +查询大漠平台账户余额。 + +- **命令**: `大漠余额` 或 `余额查询`。 +- **限制**: 仅特定用户可用，需输入验证码。 + +### 11. danding_api - 管理 API + +供超级用户使用的后台管理功能。 + +- **主要命令**: + - `在线人数`: 查询当前活跃用户。 + - `生成卡密 <类型>`: 生成天/周/月卡。 + - `用户加时 <用户名> <类型>`: 直接为特定用户增加时长。 + +### 12. danding_help - 帮助菜单 + +系统的官方指引手册。 + +- **主要命令**: `帮助`、`下载`、`活动`（别名：`最新活动`、`群活动`、`hd`）、`每日试用`、`正式版如何运行` 等。 +- **活动海报推送**: 发送 `活动` 时自动请求服务端接口获取当前正在进行中的运营活动，并向群内发送对应的活动宣传海报及说明。 +- **限制**: 仅在特定群 `621016172` 可用。 + +### 13. command_list - 指令列表 + +快速查阅所有可用指令。 + +- **命令**: `指令列表`。 + +### 14. danding_sign_in - 每日签到 + +独立的每日签到命令，与抽卡签到、WPF 客户端签到**共用同一份签到数据**（xapi `bot_gacha_daily_sign_in` 表，每用户每天仅一次，任一入口先到先得）。 + +- **主要命令**: `签到`（别名：`每日签到`、`打卡`）。 +- **积分奖励**: 随机 1-100 积分（与抽卡签到一致），积分流水 `source=gacha_sign`、`reason=抽卡签到`。 +- **权限**: 白名单群聊 `621016172` 或管理员可用；私聊非管理员不可用。 +- **数据共用说明**: 同一天内，抽卡/WPF/命令签到任一入口先触发者获得积分，后触发者提示"今天已签到"，不会重复发放。 + +--- + +## 常见问题 (FAQ) + +- **Q: 为什么某些命令没反应？** + A: 部分插件（如 `danding_help`）限制了特定群聊使用；管理指令需要配置 `SUPERUSERS`。 +- **Q: 积分有什么用？** + A: 目前主要用于赛马下注及展示排名。 +- **Q: 抽卡奖励如何领取？** + A: 抽中 SSR/SP 或解锁特定成就后，请截屏联系管理员。 \ No newline at end of file diff --git a/danding_bot/plugins/danding_qqpush/README.md b/danding_bot/plugins/danding_qqpush/README.md index 841c774..4f0e7d4 100644 --- a/danding_bot/plugins/danding_qqpush/README.md +++ b/danding_bot/plugins/danding_qqpush/README.md @@ -6,10 +6,12 @@ - ✅ 通过 HTTP API 推送消息到指定 QQ 群 - ✅ 自动将文本渲染为图片，避免长文本刷屏 +- ✅ 统一通知卡片版式，群聊/私聊使用不同场景徽章 - ✅ 可选 @指定 QQ 用户；省略 `qq` 时直接发送群消息 - ✅ 使用 `#` 符号表示换行 - ✅ 基于 Token 的简单鉴权机制 - ✅ 支持中文文本渲染 +- ✅ 自带超级用户预览命令「测试通知图片」，可在群聊/私聊中直接核对卡片版式 ## 目录结构 @@ -19,11 +21,70 @@ danding_qqpush/ ├── config.py # 配置模块 ├── api.py # FastAPI 路由模块 ├── text_parser.py # 文本处理模块 -├── image_render.py # 图片生成模块 +├── image_render.py # 通知卡片生成模块 +├── article_render.py # 公告 HTML 截图模块 +├── preview_image_render.py # 预览命令共用的渲染器实例 ├── sender.py # 消息发送模块 +├── test_command.py # 超级用户「测试通知图片」预览命令 └── utils.py # 工具函数模块 ``` +## 通知卡片样式 + +群聊与私聊的普通图片通知（即未传 `html` 的文本通知）共用同一套「冷灰外衬 + 白色圆角卡片」版式，只通过右上角的场景徽章区分来源： + +- 顶部为蓝色品牌头部：左侧固定文案「蛋定助手」，右侧为白底胶囊场景徽章 +- 正文使用比纯黑更柔和的墨色，行距不低于 1.6 倍 +- 底部为分割线与脚注「本消息由蛋定助手自动生成」 +- 正文按宽度自动折行，并做中文避头尾处理 +- Emoji 按整体方块宽度测量，避免溢出卡片；Emoji 数据源不可用时自动降级为普通文本绘制，通知仍会正常发出 + +### 场景徽章 + +| 场景 (`scene`) | 徽章文案 | 徽章颜色 | 使用位置 | +| -------------- | -------- | -------- | -------- | +| `common` | 通用通知 | 灰色 | 渲染默认值 | +| `group` | 群聊通知 | 品牌蓝 | `POST /danding/qqpush/{token}` 的普通文本通知 | +| `private` | 私聊通知 | 橙色 | `POST /danding/private/{token}` 的图片通知 | + +- `scene` 是渲染层内部参数，**不新增配置项、也不新增 API 请求字段**：请求体格式保持不变，场景由插件按接口自动选择（群聊接口传 `group`，私聊接口传 `private`） +- 未知的 `scene` 取值会回退为 `common` +- 传入 `html` 的公告推送使用 Chromium 截图（`article_render.py`），不使用本卡片版式 + +```python +renderer.render(text) # 通用通知 +renderer.render(text, scene="group") # 群聊通知 +renderer.render(text, scene="private") # 私聊通知 +renderer.render_to_base64(text, title="🏇 赛马开赛", scene="group") # 自定义标题行 +renderer.render(text, title=None) # 不绘制标题行 +``` + +`title` 默认沿用旧版固定文案「蛋定助手通知您：」以保持向后兼容；显式传 `None` 或空字符串才隐藏标题行（卡片头部已展示「蛋定助手」品牌，需要更紧凑的版式时可以隐藏）。 + +## 测试通知图片命令 + +不想注册 HTTP 接口、也不想占用群聊刷屏时，可以直接用机器人自带的预览命令核对卡片版式。 + +| 项目 | 说明 | +| ---- | ---- | +| 主命令 | `测试通知图片` | +| 别名 | `通知图片测试`、`测试通知卡片` | +| 权限 | 仅 NoneBot 超级用户（`SUPERUSER`，即 `.env` 的 `SUPERUSERS`） | +| 触发方式 | 群聊直接发送，或与机器人私聊发送；均不需要 @机器人 | + +- 群聊中触发时渲染 `scene="group"`（群聊通知徽章），私聊中触发时渲染 `scene="private"`（私聊通知徽章） +- 预览内容是固定示例，覆盖标题行、中文、多行、Emoji 与中文标点；标题行特意写成 `# 会被解析为换行：`，同时演示 `#` 作为换行符的用法 +- 命令复用通知卡片渲染器与插件配置（字体、宽度、行距等），不新增配置项，也不改变任何 HTTP 接口 +- 渲染是同步阻塞操作，命令会把它放到 `asyncio.to_thread` 执行，避免阻塞事件循环 +- 渲染失败时机器人会记录 `[Danding_QqPush] 通知图片测试失败` 日志，并回复「通知图片生成失败，请查看机器人日志后重试。」 + +### 用法示例 + +``` +（群聊 / 私聊中直接发送） +测试通知图片 +``` + ## 安装与配置 ### 1. 确保依赖已安装 @@ -167,6 +228,13 @@ class Config(BaseModel): """字体文件路径列表""" ``` +与通知卡片相关的渲染约定（沿用上述配置，无需新增配置项）： + +- `ImageBgColor` 用作卡片填充色；卡片外衬固定为冷灰色，保证浅色卡片轮廓可见 +- `ImageTextColor` 传入纯黑 `(0, 0, 0)` 时会自动替换为柔和的 `(31, 35, 41)`，其他颜色原样使用 +- `ImageLineSpacing` 低于 `1.6` 时正文按 `1.6` 倍行距渲染 +- `ImageWidth`/`ImagePadding`/`FontPaths` 的语义不变，卡片与正文均按它们计算版式 + ## 文本格式说明 - 使用 `#` 符号表示换行 @@ -206,6 +274,13 @@ class Config(BaseModel): - 检查 Bot 是否有发送消息的权限 - 查看日志中的详细错误信息 +### 问题：发送「测试通知图片」没有反应 + +**解决方案**: +- 确认发送者的 QQ 号在 `.env` 的 `SUPERUSERS` 中（该命令仅超级用户可用，其他用户发送不会触发、也不会有提示） +- 确认消息就是命令本身（`测试通知图片` / `通知图片测试` / `测试通知卡片`），命令后跟其他内容不会触发 +- 若机器人回复「通知图片生成失败，请查看机器人日志后重试。」，按日志中 `[Danding_QqPush] 通知图片测试失败` 的报错排查字体路径与配置 + ## 测试 使用提供的测试脚本进行测试： @@ -215,3 +290,9 @@ python test_qqpush.py ``` 修改脚本中的 `group_id` 和 `qq` 为实际值。 + +插件自带的自动化测试（不需要真实 QQ 连接）： + +```bash +python -m pytest tests/test_danding_qqpush_test_command.py -q +``` diff --git a/danding_bot/plugins/danding_qqpush/__init__.py b/danding_bot/plugins/danding_qqpush/__init__.py index 1e86f18..3419e86 100644 --- a/danding_bot/plugins/danding_qqpush/__init__.py +++ b/danding_bot/plugins/danding_qqpush/__init__.py @@ -1,65 +1,74 @@ -"""Danding_QqPush 插件初始化模块""" -from nonebot import get_driver -from nonebot.log import logger -from nonebot.plugin import PluginMetadata - -from .config import Config -from .api import create_routes -from .sender import sender - - -__plugin_meta__ = PluginMetadata( - name="danding_qqpush", - description="通过外部 HTTP API 向 QQ 群定向推送通知", - usage=""" - API 接口： - POST /danding/qqpush/{token} - - 请求参数： - { - "group_id": 123456789, +"""Danding_QqPush 插件初始化模块""" +from nonebot import get_driver +from nonebot.log import logger +from nonebot.plugin import PluginMetadata + +from .config import Config +from .api import create_routes +from .sender import sender + + +__plugin_meta__ = PluginMetadata( + name="danding_qqpush", + description="通过外部 HTTP API 向 QQ 群定向推送通知", + usage=""" + API 接口： + POST /danding/qqpush/{token} + + 请求参数： + { + "group_id": 123456789, "text": "系统告警#数据库连接失败#请立即处理" - } - - 说明： - - text 中的 # 表示换行 - - 消息会自动渲染为图片并发送到指定群 - """, - config=Config, -) - - -# 加载配置 -plugin_config = Config.model_validate(get_driver().config.model_dump()) - - -def register_routes(): - """注册 FastAPI 路由""" - driver = get_driver() - - # 创建并注册路由 - routes = create_routes(plugin_config.Token, plugin_config) - driver.server_app.include_router(routes) - - masked = plugin_config.Token[:4] + "***" if len(plugin_config.Token) > 4 else "***" - logger.info(f"[Danding_QqPush] API 路由已注册: /danding/qqpush/{masked}") - - -# 插件加载时注册路由 -try: - register_routes() - logger.info("[Danding_QqPush] 插件加载成功") -except Exception as e: - logger.error(f"[Danding_QqPush] 插件加载失败: {str(e)}") - - -# Bot 连接时自动初始化 sender -driver = get_driver() -@driver.on_bot_connect -async def _(bot): - """Bot 连接时自动设置 sender""" - try: - sender.set_bot(bot) - logger.info(f"[Danding_QqPush] Bot 已连接: {bot.self_id}") - except Exception as e: - logger.error(f"[Danding_QqPush] Bot 连接初始化失败: {e}") + } + + 说明： + - text 中的 # 表示换行 + - 消息会自动渲染为图片并发送到指定群 + + 测试命令（仅超级用户）： + 测试通知图片（别名：通知图片测试、测试通知卡片） + - 群聊中发送渲染群聊通知卡片，私聊中发送渲染私聊通知卡片 + - 固定预览内容覆盖标题、中文、多行、Emoji 与中文标点，用于肉眼核对版式 + """, + config=Config, +) + + +# 加载配置 +plugin_config = Config.model_validate(get_driver().config.model_dump()) + + +# 配置初始化完成后再导入测试命令模块，避免反向依赖尚未初始化完成的插件包 +from . import test_command # noqa: E402,F401 - 导入即注册测试命令 matcher + + +def register_routes(): + """注册 FastAPI 路由""" + driver = get_driver() + + # 创建并注册路由 + routes = create_routes(plugin_config.Token, plugin_config) + driver.server_app.include_router(routes) + + masked = plugin_config.Token[:4] + "***" if len(plugin_config.Token) > 4 else "***" + logger.info(f"[Danding_QqPush] API 路由已注册: /danding/qqpush/{masked}") + + +# 插件加载时注册路由 +try: + register_routes() + logger.info("[Danding_QqPush] 插件加载成功") +except Exception as e: + logger.error(f"[Danding_QqPush] 插件加载失败: {str(e)}") + + +# Bot 连接时自动初始化 sender +driver = get_driver() +@driver.on_bot_connect +async def _(bot): + """Bot 连接时自动设置 sender""" + try: + sender.set_bot(bot) + logger.info(f"[Danding_QqPush] Bot 已连接: {bot.self_id}") + except Exception as e: + logger.error(f"[Danding_QqPush] Bot 连接初始化失败: {e}") \ No newline at end of file diff --git a/danding_bot/plugins/danding_qqpush/api.py b/danding_bot/plugins/danding_qqpush/api.py index a6cafbc..0191528 100644 --- a/danding_bot/plugins/danding_qqpush/api.py +++ b/danding_bot/plugins/danding_qqpush/api.py @@ -1,10 +1,10 @@ -"""API 接口模块 - FastAPI 路由定义""" -from fastapi import APIRouter, Request, HTTPException -from pydantic import BaseModel -import asyncio -from typing import Optional +"""API 接口模块 - FastAPI 路由定义""" +from fastapi import APIRouter, Request, HTTPException +from pydantic import BaseModel +import asyncio +from typing import Optional from nonebot import logger - + from .config import Config from .article_render import render_article_to_base64 from .text_parser import TextParser @@ -14,32 +14,32 @@ from .utils import validate_token # Module-level singleton: load font once, reuse across requests _renderer: Optional['ImageRenderer'] = None - -def _get_renderer(config: Config) -> 'ImageRenderer': - global _renderer - if _renderer is None: - _renderer = ImageRenderer( - width=config.ImageWidth, - font_size=config.ImageFontSize, - padding=config.ImagePadding, - line_spacing=config.ImageLineSpacing, - bg_color=config.ImageBgColor, - text_color=config.ImageTextColor, - font_paths=config.FontPaths, - ) - return _renderer - - - -# 请求体模型 -class PushRequest(BaseModel): - """推送请求模型""" - group_id: int - """接收消息的 QQ 群号""" - + +def _get_renderer(config: Config) -> 'ImageRenderer': + global _renderer + if _renderer is None: + _renderer = ImageRenderer( + width=config.ImageWidth, + font_size=config.ImageFontSize, + padding=config.ImagePadding, + line_spacing=config.ImageLineSpacing, + bg_color=config.ImageBgColor, + text_color=config.ImageTextColor, + font_paths=config.FontPaths, + ) + return _renderer + + + +# 请求体模型 +class PushRequest(BaseModel): + """推送请求模型""" + group_id: int + """接收消息的 QQ 群号""" + qq: Optional[int] = None """可选的被 @ QQ 号；未传时仅发送群消息""" - + text: str """通知文本（# 表示换行）""" @@ -51,67 +51,67 @@ class PushRequest(BaseModel): published_at: Optional[str] = None """公告发布时间""" - - -class PrivatePushRequest(BaseModel): - """私聊推送请求模型""" - user_id: int - """接收私聊消息的好友 QQ 号""" - - text: Optional[str] = None - """通知文本（# 表示换行）""" - - text_only: bool = False - """True=发送纯文本；False（默认）=渲染为图片发送""" - - html: Optional[str] = None - """可选的完整公告 HTML；传入时使用虚拟浏览器截图""" - - title: Optional[str] = None - """公告标题""" - - published_at: Optional[str] = None - """公告发布时间""" - - - -# 响应模型 -class PushResponse(BaseModel): - """推送响应模型""" - success: bool - """是否成功""" - - message: str - """响应消息""" - - data: Optional[dict] = None - """返回数据（如有）""" - - -# 创建路由器 -router = APIRouter() - - -def create_routes(token: str, config: Config): - """ - 创建 API 路由 - - Args: - token: 鉴权 Token - config: 配置对象 - """ - + + +class PrivatePushRequest(BaseModel): + """私聊推送请求模型""" + user_id: int + """接收私聊消息的好友 QQ 号""" + + text: Optional[str] = None + """通知文本（# 表示换行）""" + + text_only: bool = False + """True=发送纯文本；False（默认）=渲染为图片发送""" + + html: Optional[str] = None + """可选的完整公告 HTML；传入时使用虚拟浏览器截图""" + + title: Optional[str] = None + """公告标题""" + + published_at: Optional[str] = None + """公告发布时间""" + + + +# 响应模型 +class PushResponse(BaseModel): + """推送响应模型""" + success: bool + """是否成功""" + + message: str + """响应消息""" + + data: Optional[dict] = None + """返回数据（如有）""" + + +# 创建路由器 +router = APIRouter() + + +def create_routes(token: str, config: Config): + """ + 创建 API 路由 + + Args: + token: 鉴权 Token + config: 配置对象 + """ + @router.post("/danding/qqpush/{request_token}", response_model=PushResponse) async def qqpush(request_token: str, request: Request, data: PushRequest): - """ - QQ 消息推送接口 - - Args: - request: FastAPI 请求对象 - data: 推送请求数据 - - Returns: - 推送结果 + """ + QQ 消息推送接口 + + Args: + request: FastAPI 请求对象 + data: 推送请求数据 + + Returns: + 推送结果 """ try: if not validate_token(request_token, token): @@ -120,27 +120,27 @@ def create_routes(token: str, config: Config): # 1. 验证参数 if not data.group_id: raise HTTPException(status_code=400, detail="group_id 不能为空") - + if not data.text or not isinstance(data.text, str): - raise HTTPException(status_code=400, detail="text 不能为空且必须是字符串") - - # 2. 检查 Bot 是否在线 - bot = sender.get_bot() - if not bot: - logger.error("Bot 实例未设置，无法发送消息") - raise HTTPException( - status_code=500, - detail="Bot 未连接，请检查机器人状态" - ) - - # 3. 文本处理 - text_parser = TextParser(max_length=config.MaxTextLength) - if not text_parser.validate_text(data.text): - raise HTTPException(status_code=400, detail="文本内容无效") - - parsed_text = text_parser.parse(data.text) - logger.info(f"解析文本: {parsed_text[:50]}..." if len(parsed_text) > 50 else parsed_text) - + raise HTTPException(status_code=400, detail="text 不能为空且必须是字符串") + + # 2. 检查 Bot 是否在线 + bot = sender.get_bot() + if not bot: + logger.error("Bot 实例未设置，无法发送消息") + raise HTTPException( + status_code=500, + detail="Bot 未连接，请检查机器人状态" + ) + + # 3. 文本处理 + text_parser = TextParser(max_length=config.MaxTextLength) + if not text_parser.validate_text(data.text): + raise HTTPException(status_code=400, detail="文本内容无效") + + parsed_text = text_parser.parse(data.text) + logger.info(f"解析文本: {parsed_text[:50]}..." if len(parsed_text) > 50 else parsed_text) + # 4. 原始公告使用 Chromium 截图，普通通知保持既有 Pillow 渲染。 if data.html: image_base64 = await render_article_to_base64( @@ -154,128 +154,132 @@ def create_routes(token: str, config: Config): qq=data.qq, ) else: - image_base64 = await asyncio.to_thread(_get_renderer(config).render_to_base64, parsed_text) + image_base64 = await asyncio.to_thread( + _get_renderer(config).render_to_base64, parsed_text, scene="group" + ) send_result = await sender.send_to_group( group_id=data.group_id, image_base64=image_base64, qq=data.qq, ) logger.info("图片生成成功") - - if not send_result["success"]: - logger.error(f"消息发送失败: {send_result['error']}") - raise HTTPException( - status_code=500, - detail=send_result["message"] - ) - - logger.info(f"消息发送成功 - 群: {data.group_id}, @: {data.qq}") - - return PushResponse( - success=True, - message="推送成功", - data={ - "group_id": data.group_id, - "qq": data.qq, - "message_id": send_result["data"].get("message_id") - } - ) - - except HTTPException: - raise - - except Exception as e: - logger.exception(f"推送接口异常: {e}") - raise HTTPException( - status_code=500, - detail="服务器内部错误" - ) - - - @router.post("/danding/private/{request_token}", response_model=PushResponse) - async def private_push(request_token: str, request: Request, data: PrivatePushRequest): - """ - 好友私聊推送接口 - - 仅可向好友发送私聊消息；目标不在好友列表或发送失败时返回错误。 - """ - try: - if not validate_token(request_token, token): - raise HTTPException(status_code=403, detail="Token 验证失败") - - # 1. 验证参数 - if not data.user_id: - raise HTTPException(status_code=400, detail="user_id 不能为空") - - if not (data.text or data.html): - raise HTTPException(status_code=400, detail="text 和 html 不能同时为空") - - # 2. 检查 Bot 是否在线 - bot = sender.get_bot() - if not bot: - logger.error("Bot 实例未设置，无法发送消息") - raise HTTPException(status_code=500, detail="Bot 未连接，请检查机器人状态") - - # 3. 检查目标是否为好友 - try: - is_friend = await sender.is_friend(data.user_id) - except Exception as e: - logger.error(f"查询好友列表失败: {e}") - raise HTTPException(status_code=500, detail=f"查询好友列表失败: {e}") - - if not is_friend: - logger.warning(f"目标用户不是好友: {data.user_id}") - raise HTTPException( - status_code=400, - detail=f"用户 {data.user_id} 不在好友列表中，无法发送私聊消息" - ) - - # 4. 发送消息 - if data.text_only and data.text: - text_parser = TextParser(max_length=config.MaxTextLength) - if not text_parser.validate_text(data.text): - raise HTTPException(status_code=400, detail="文本内容无效") - parsed_text = text_parser.parse(data.text) - send_result = await sender.send_text_to_private(data.user_id, parsed_text) - elif data.html: - image_base64 = await render_article_to_base64( - data.html, - data.title or "蛋定助手通知", - data.published_at, - ) - send_result = await sender.send_to_private(data.user_id, image_base64) - else: - text_parser = TextParser(max_length=config.MaxTextLength) - if not text_parser.validate_text(data.text): - raise HTTPException(status_code=400, detail="文本内容无效") - parsed_text = text_parser.parse(data.text) - image_base64 = await asyncio.to_thread(_get_renderer(config).render_to_base64, parsed_text) - send_result = await sender.send_to_private(data.user_id, image_base64) - - logger.info("私聊文本已准备发送" if data.text_only else "私聊图片生成成功") - - if not send_result["success"]: - logger.error(f"私聊消息发送失败: {send_result.get('error')}") - raise HTTPException(status_code=500, detail=send_result["message"]) - - logger.info(f"私聊消息发送成功 - 好友: {data.user_id}") - - return PushResponse( - success=True, - message="私聊推送成功", - data={ - "user_id": data.user_id, - "message_id": send_result["data"].get("message_id") - } - ) - - except HTTPException: - raise - - except Exception as e: - logger.exception(f"私聊推送接口异常: {e}") - raise HTTPException( - status_code=500, - detail="服务器内部错误" - ) - return router + + if not send_result["success"]: + logger.error(f"消息发送失败: {send_result['error']}") + raise HTTPException( + status_code=500, + detail=send_result["message"] + ) + + logger.info(f"消息发送成功 - 群: {data.group_id}, @: {data.qq}") + + return PushResponse( + success=True, + message="推送成功", + data={ + "group_id": data.group_id, + "qq": data.qq, + "message_id": send_result["data"].get("message_id") + } + ) + + except HTTPException: + raise + + except Exception as e: + logger.exception(f"推送接口异常: {e}") + raise HTTPException( + status_code=500, + detail="服务器内部错误" + ) + + + @router.post("/danding/private/{request_token}", response_model=PushResponse) + async def private_push(request_token: str, request: Request, data: PrivatePushRequest): + """ + 好友私聊推送接口 + + 仅可向好友发送私聊消息；目标不在好友列表或发送失败时返回错误。 + """ + try: + if not validate_token(request_token, token): + raise HTTPException(status_code=403, detail="Token 验证失败") + + # 1. 验证参数 + if not data.user_id: + raise HTTPException(status_code=400, detail="user_id 不能为空") + + if not (data.text or data.html): + raise HTTPException(status_code=400, detail="text 和 html 不能同时为空") + + # 2. 检查 Bot 是否在线 + bot = sender.get_bot() + if not bot: + logger.error("Bot 实例未设置，无法发送消息") + raise HTTPException(status_code=500, detail="Bot 未连接，请检查机器人状态") + + # 3. 检查目标是否为好友 + try: + is_friend = await sender.is_friend(data.user_id) + except Exception as e: + logger.error(f"查询好友列表失败: {e}") + raise HTTPException(status_code=500, detail=f"查询好友列表失败: {e}") + + if not is_friend: + logger.warning(f"目标用户不是好友: {data.user_id}") + raise HTTPException( + status_code=400, + detail=f"用户 {data.user_id} 不在好友列表中，无法发送私聊消息" + ) + + # 4. 发送消息 + if data.text_only and data.text: + text_parser = TextParser(max_length=config.MaxTextLength) + if not text_parser.validate_text(data.text): + raise HTTPException(status_code=400, detail="文本内容无效") + parsed_text = text_parser.parse(data.text) + send_result = await sender.send_text_to_private(data.user_id, parsed_text) + elif data.html: + image_base64 = await render_article_to_base64( + data.html, + data.title or "蛋定助手通知", + data.published_at, + ) + send_result = await sender.send_to_private(data.user_id, image_base64) + else: + text_parser = TextParser(max_length=config.MaxTextLength) + if not text_parser.validate_text(data.text): + raise HTTPException(status_code=400, detail="文本内容无效") + parsed_text = text_parser.parse(data.text) + image_base64 = await asyncio.to_thread( + _get_renderer(config).render_to_base64, parsed_text, scene="private" + ) + send_result = await sender.send_to_private(data.user_id, image_base64) + + logger.info("私聊文本已准备发送" if data.text_only else "私聊图片生成成功") + + if not send_result["success"]: + logger.error(f"私聊消息发送失败: {send_result.get('error')}") + raise HTTPException(status_code=500, detail=send_result["message"]) + + logger.info(f"私聊消息发送成功 - 好友: {data.user_id}") + + return PushResponse( + success=True, + message="私聊推送成功", + data={ + "user_id": data.user_id, + "message_id": send_result["data"].get("message_id") + } + ) + + except HTTPException: + raise + + except Exception as e: + logger.exception(f"私聊推送接口异常: {e}") + raise HTTPException( + status_code=500, + detail="服务器内部错误" + ) + return router diff --git a/danding_bot/plugins/danding_qqpush/image_render.py b/danding_bot/plugins/danding_qqpush/image_render.py index 0f61c24..90b64f7 100644 --- a/danding_bot/plugins/danding_qqpush/image_render.py +++ b/danding_bot/plugins/danding_qqpush/image_render.py @@ -1,215 +1,473 @@ -"""图片生成模块 - 将文本渲染为图片""" -from PIL import Image, ImageDraw, ImageFont -from pilmoji import Pilmoji -import io -import base64 -from typing import Tuple - - -class ImageRenderer: - """图片渲染器""" - - DEFAULT_FONT_PATHS = [ - "/usr/share/fonts/opentype/noto/NotoSansCJK-Regular.ttc", - "/usr/share/fonts/truetype/noto/NotoColorEmoji.ttf", - "C:/Windows/Fonts/msyh.ttc", - "C:/Windows/Fonts/simhei.ttf", - "C:/Windows/Fonts/seguiemj.ttf", - ] - - def __init__( - self, - width: int = 800, - font_size: int = 24, - padding: int = 30, - line_spacing: float = 1.4, - bg_color: Tuple[int, int, int] = (252, 252, 252), - text_color: Tuple[int, int, int] = (0, 0, 0), - font_paths: list = None - ): - """ - 初始化图片渲染器 - - Args: - width: 图片宽度 - font_size: 字体大小 - padding: 内边距 - line_spacing: 行距倍数 - bg_color: 背景颜色 (R, G, B) - text_color: 文本颜色 (R, G, B) - font_paths: 字体文件路径列表 - """ - self.width = width - self.font_size = font_size - self.padding = padding - self.line_spacing = line_spacing - self.bg_color = bg_color - self.text_color = text_color - self.font_paths = font_paths or self.DEFAULT_FONT_PATHS.copy() - - # 加载字体 - self.font = self._load_font() - - def _load_font(self) -> ImageFont.FreeTypeFont: - """加载字体""" - return self._load_font_by_size(self.font_size) - - def _load_font_by_size(self, font_size: int) -> ImageFont.FreeTypeFont: - """ - 加载指定大小的字体 - - Args: - font_size: 字体大小 - - Returns: - 字体对象 - """ - for font_path in self.font_paths: - try: - font = ImageFont.truetype(font_path, font_size) - return font - except Exception: - continue - - # 如果都失败了，使用默认字体 - return ImageFont.load_default() - - def _calculate_text_dimensions(self, text: str) -> Tuple[int, int]: - """ - 计算文本的尺寸（使用缓存的 Draw 对象，避免重复创建临时 Image） - - Args: - text: 文本内容 - - Returns: - (宽度, 高度) - """ - if not hasattr(self, '_cached_draw'): - self._cached_img = Image.new('RGB', (1, 1), color=self.bg_color) - self._cached_draw = ImageDraw.Draw(self._cached_img) - - bbox = self._cached_draw.textbbox((0, 0), text, font=self.font) - width = bbox[2] - bbox[0] - height = bbox[3] - bbox[1] - - return width, height - - def _smart_text_wrap(self, text: str, max_width: int) -> list: - """ - 智能文本换行（使用 font.getlength 避免逐字符创建临时 Image） - - Args: - text: 文本内容 - max_width: 最大宽度 - - Returns: - 换行后的文本列表 - """ - lines = [] - current_line = "" - current_width = 0.0 - - for char in text: - char_width = self.font.getlength(char) - - if current_width + char_width <= max_width: - current_line += char - current_width += char_width - else: - if current_line: - lines.append(current_line) - current_line = char - current_width = char_width - - if current_line: - lines.append(current_line) - - return lines - - def render(self, text: str, title: str = "蛋定助手通知您：") -> bytes: - """ - 将文本渲染为图片 - - Args: - text: 文本内容（已处理换行符） - - Returns: - 图片的字节数据 - """ - if not text: - text = "（空消息）" - - # 计算有效宽度 - effective_width = self.width - (2 * self.padding) - - # 按段落处理 - all_lines = [] - for paragraph in text.split('\n'): - if not paragraph: - all_lines.append("") - continue - - wrapped_lines = self._smart_text_wrap(paragraph, effective_width) - all_lines.extend(wrapped_lines) - - # 计算行高 - _, line_height = self._calculate_text_dimensions("测试") - total_line_height = int(line_height * self.line_spacing) - - # 标题相关 - title_font_size = int(self.font_size * 1.3) # 标题字体放大1.3倍 - title_font = self._load_font_by_size(title_font_size) - title_height = int(line_height * 1.5) # 标题区域高度 - divider_height = 2 # 分割线高度 - divider_spacing = 10 # 分割线与标题的间距 - - # 计算总高度（包含标题区域） - content_height = (len(all_lines) * total_line_height) + (2 * self.padding) - header_height = title_height + divider_spacing + divider_height + self.padding - total_height = content_height + header_height - height = max(total_height, 200) # 最小高度 - - # 创建图片 - image = Image.new('RGB', (self.width, height), color=self.bg_color) - draw = ImageDraw.Draw(image) - - title_x = self.padding - title_y = self.padding - with Pilmoji(image) as pilmoji: - pilmoji.text((title_x, title_y), title, font=title_font, fill=self.text_color) - - # 绘制分割线 - divider_y = title_y + title_height + 10 - draw.line( - [(self.padding, divider_y), (self.width - self.padding, divider_y)], - fill=(200, 200, 200), - width=divider_height - ) - - y = divider_y + divider_spacing + self.padding - with Pilmoji(image) as pilmoji: - for line in all_lines: - if line: - pilmoji.text((self.padding, y), line, font=self.font, fill=self.text_color) - y += total_line_height - - # 转换为字节流 - img_byte_arr = io.BytesIO() - image.save(img_byte_arr, format='PNG') - img_byte_arr.seek(0) - - return img_byte_arr.getvalue() - - def render_to_base64(self, text: str, title: str = "蛋定助手通知您：") -> str: - """ - 将文本渲染为图片并返回 base64 编码 - - Args: - text: 文本内容 - - Returns: - base64 编码的图片数据 - """ - img_bytes = self.render(text, title=title) - base64_str = base64.b64encode(img_bytes).decode('utf-8') - return f"base64://{base64_str}" +"""图片生成模块 - 将文本渲染为统一的通知卡片图片""" +import base64 +import io +import logging +from contextlib import ExitStack +from typing import Dict, Iterator, List, Optional, Tuple + +from PIL import Image, ImageDraw, ImageFont +from pilmoji import Pilmoji + +try: # 复用 Pilmoji 的 Emoji 识别正则，让 Emoji 序列作为整体参与测量与换行 + from pilmoji.helpers import EMOJI_REGEX +except Exception: # pragma: no cover - 依赖内部结构变化时的兜底 + EMOJI_REGEX = None + +logger = logging.getLogger(__name__) + + +class ImageRenderer: + """通知卡片渲染器：冷灰外衬 + 白色圆角卡片 + 品牌头部 + 场景徽章。""" + + DEFAULT_FONT_PATHS = [ + "/usr/share/fonts/opentype/noto/NotoSansCJK-Regular.ttc", + "/usr/share/fonts/truetype/noto/NotoColorEmoji.ttf", + "C:/Windows/Fonts/msyh.ttc", + "C:/Windows/Fonts/simhei.ttf", + "C:/Windows/Fonts/seguiemj.ttf", + ] + + # 版式 + OUTER_MARGIN = 24 + """卡片与画布边缘的间距（冷灰外衬宽度）""" + CARD_RADIUS = 18 + """卡片圆角半径""" + MIN_HEIGHT = 200 + """图片最小高度""" + BODY_LINE_SPACING_MIN = 1.6 + """正文字段最小行距倍数，保证长文本的可读性""" + + # 配色 + OUTER_BG_COLOR: Tuple[int, int, int] = (237, 240, 245) + """冷灰外衬颜色""" + BRAND_COLOR: Tuple[int, int, int] = (22, 119, 255) + """品牌头部蓝色""" + BRAND_TEXT_COLOR: Tuple[int, int, int] = (255, 255, 255) + """品牌标题文字颜色""" + DEFAULT_TEXT_COLOR: Tuple[int, int, int] = (31, 35, 41) + """正文默认墨色，避免纯黑在小字号下的生硬对比""" + DIVIDER_COLOR: Tuple[int, int, int] = (229, 230, 235) + """分割线颜色""" + FOOTER_TEXT_COLOR: Tuple[int, int, int] = (134, 144, 156) + """脚注文字颜色""" + BADGE_BG_COLOR: Tuple[int, int, int] = (255, 255, 255) + """场景徽章底色""" + + BRAND_LABEL = "蛋定助手" + """品牌头部固定文案""" + FOOTER_TEXT = "本消息由蛋定助手自动生成" + """底部脚注文案""" + + DEFAULT_TITLE = "蛋定助手通知您：" + """默认标题文案（沿用旧版行为以保证向后兼容）；显式传 None 或空字符串则不绘制标题行""" + + DEFAULT_SCENE = "common" + """默认场景标识""" + SCENE_STYLES: Dict[str, Dict[str, object]] = { + "common": {"label": "通用通知", "color": (78, 89, 105)}, + "group": {"label": "群聊通知", "color": (22, 119, 255)}, + "private": {"label": "私聊通知", "color": (250, 84, 28)}, + } + """场景徽章样式表（未知场景回退为 common）""" + + # 中文避头尾：行首禁则与行尾禁则 + LINE_START_FORBIDDEN = "，。、；：！？）】》」』”’…·,.;:!?)]}" + LINE_END_FORBIDDEN = "（【《「『“‘([{" + + def __init__( + self, + width: int = 800, + font_size: int = 24, + padding: int = 30, + line_spacing: float = 1.4, + bg_color: Tuple[int, int, int] = (252, 252, 252), + text_color: Tuple[int, int, int] = (0, 0, 0), + font_paths: list = None + ): + """ + 初始化图片渲染器 + + Args: + width: 图片宽度 + font_size: 字体大小 + padding: 卡片内边距 + line_spacing: 行距倍数 + bg_color: 卡片底色 (R, G, B) + text_color: 正文字色 (R, G, B)；纯黑会回退为柔和墨色 + font_paths: 字体文件路径列表 + """ + self.width = width + self.font_size = font_size + self.padding = padding + self.line_spacing = line_spacing + self.bg_color = bg_color + # 浅色卡片上的纯黑过于生硬，统一替换为墨色 + self.text_color = self.DEFAULT_TEXT_COLOR if tuple(text_color) == (0, 0, 0) else tuple(text_color) + self.font_paths = font_paths or self.DEFAULT_FONT_PATHS.copy() + + self._width_cache: Dict[tuple, float] = {} + self._measuring_image: Optional[Image.Image] = None + self._measuring_draw: Optional[ImageDraw.ImageDraw] = None + # Emoji 由 Pilmoji 从远程源获取，一旦失败就持续降级为普通绘制，避免每条通知都卡在超时上 + self._emoji_enabled = True + + # 加载字体 + self.font = self._load_font() + + def _load_font(self) -> ImageFont.FreeTypeFont: + """加载字体""" + return self._load_font_by_size(self.font_size) + + def _load_font_by_size(self, font_size: int) -> ImageFont.FreeTypeFont: + """ + 加载指定大小的字体 + + Args: + font_size: 字体大小 + + Returns: + 字体对象 + """ + for font_path in self.font_paths: + try: + font = ImageFont.truetype(font_path, font_size) + return font + except Exception: + continue + + # 如果都失败了，使用默认字体 + return ImageFont.load_default() + + def _get_measuring_draw(self) -> ImageDraw.ImageDraw: + """复用测量用 Draw 对象，避免重复创建临时 Image。""" + if self._measuring_draw is None: + self._measuring_image = Image.new('RGB', (1, 1), color=(255, 255, 255)) + self._measuring_draw = ImageDraw.Draw(self._measuring_image) + return self._measuring_draw + + def _calculate_text_dimensions(self, text: str) -> Tuple[int, int]: + """ + 计算文本的尺寸 + + Args: + text: 文本内容 + + Returns: + (宽度, 高度) + """ + return self._text_size(text, self.font) + + def _text_size(self, text: str, font: ImageFont.FreeTypeFont) -> Tuple[int, int]: + """ + 计算指定字体下文本的尺寸 + + Args: + text: 文本内容 + font: 字体对象 + + Returns: + (宽度, 高度) + """ + draw = self._get_measuring_draw() + bbox = draw.textbbox((0, 0), text, font=font) + return bbox[2] - bbox[0], bbox[3] - bbox[1] + + def _char_width(self, char: str, font: ImageFont.FreeTypeFont) -> float: + """ + 测量单个字符宽度，缺字形时按方块宽度兜底。 + + Args: + char: 单个字符 + font: 字体对象 + + Returns: + 字符宽度（像素） + """ + key = (getattr(font, "path", None), getattr(font, "size", self.font_size), char) + width = self._width_cache.get(key) + if width is None: + try: + width = float(font.getlength(char)) + except Exception: + width = 0.0 + if width <= 0.0: + # 字体缺少该字形（Emoji 常见）时按 em 方块宽度兜底，避免整行挤在一起 + width = float(getattr(font, "size", self.font_size)) + self._width_cache[key] = width + return width + + def _iter_atoms(self, text: str, font: ImageFont.FreeTypeFont) -> Iterator[Tuple[str, float]]: + """ + 把文本拆成最小换行单元：普通字符逐个，Emoji 序列整体不拆分。 + + Args: + text: 单行文本 + + Yields: + (单元文本, 单元宽度) + """ + chunks = EMOJI_REGEX.split(text) if EMOJI_REGEX is not None else [text] + for index, chunk in enumerate(chunks): + if not chunk: + continue + if index % 2: + # Pilmoji 以 font.size 的方块绘制 Emoji + yield chunk, float(getattr(font, "size", self.font_size)) + else: + for char in chunk: + yield char, self._char_width(char, font) + + def _measure_text(self, text: str, font: ImageFont.FreeTypeFont) -> float: + """ + 测量整段文本宽度（Emoji 按 Pilmoji 的绘制宽度计算）。 + + Args: + text: 文本内容 + font: 字体对象 + + Returns: + 文本宽度（像素） + """ + return sum(width for _, width in self._iter_atoms(text, font)) + + def _wrap_paragraph(self, paragraph: str, max_width: int) -> List[str]: + """ + 智能文本换行（Emoji 整体处理，并做基础的中文避头尾） + + Args: + paragraph: 段落文本（不含换行符） + max_width: 最大宽度 + + Returns: + 换行后的文本列表 + """ + lines: List[str] = [] + current_line = "" + current_width = 0.0 + + for atom, atom_width in self._iter_atoms(paragraph, self.font): + if current_line and current_width + atom_width > max_width: + if atom in self.LINE_START_FORBIDDEN and len(current_line) > 1: + # 标点不能出现在行首：把前一字符一起挪到下一行 + tail = current_line[-1] + lines.append(current_line[:-1]) + current_line = tail + atom + current_width = self._char_width(tail, self.font) + atom_width + continue + + # 开引号/开括号不能出现在行尾：把它一起挪到下一行 + head = "" + if current_line[-1] in self.LINE_END_FORBIDDEN and len(current_line) > 1: + head = current_line[-1] + current_line = current_line[:-1] + lines.append(current_line) + current_line = head + atom + current_width = (self._char_width(head, self.font) if head else 0.0) + atom_width + continue + + current_line += atom + current_width += atom_width + + if current_line: + lines.append(current_line) + + return lines + + def _smart_text_wrap(self, text: str, max_width: int) -> List[str]: + """ + 智能文本换行（保留原有方法名） + + Args: + text: 文本内容 + max_width: 最大宽度 + + Returns: + 换行后的文本列表 + """ + return self._wrap_paragraph(text, max_width) + + def _wrap_text(self, text: str, max_width: int) -> List[str]: + """ + 按显式换行符与宽度折行 + + Args: + text: 文本内容（已处理换行符） + max_width: 正文最大宽度 + + Returns: + 折行后的文本列表 + """ + all_lines: List[str] = [] + for paragraph in text.split('\n'): + if not paragraph: + all_lines.append("") + continue + all_lines.extend(self._wrap_paragraph(paragraph, max_width)) + return all_lines or [""] + + def _paint( + self, + pilmoji: Optional[Pilmoji], + draw: ImageDraw.ImageDraw, + xy: Tuple[int, int], + text: str, + font: ImageFont.FreeTypeFont, + fill: Tuple[int, int, int], + ) -> None: + """绘制单行文本；Emoji 源不可用时退化为 PIL 原生绘制，保证通知仍能发出。""" + if pilmoji is not None and self._emoji_enabled: + try: + pilmoji.text(xy, text, font=font, fill=fill) + return + except Exception as exc: + self._emoji_enabled = False + logger.warning("Emoji 渲染失败，本次降级为普通文本绘制: %s", exc) + draw.text(xy, text, font=font, fill=fill) + + def _scene_style(self, scene: str) -> Dict[str, object]: + """获取场景徽章样式，未知场景回退为默认样式。""" + return self.SCENE_STYLES.get(scene) or self.SCENE_STYLES[self.DEFAULT_SCENE] + + def render(self, text: str, title: Optional[str] = DEFAULT_TITLE, scene: str = "common") -> bytes: + """ + 将文本渲染为通知卡片图片 + + Args: + text: 文本内容（已处理换行符） + title: 标题文案，默认沿用旧版固定文案；显式传 None 或空字符串时不绘制标题行 + scene: 场景标识，支持 common/group/private，未知值回退为 common + + Returns: + 图片的字节数据 + """ + if not text: + text = "（空消息）" + + scene_style = self._scene_style(scene) + + # 版式计算 + margin = self.OUTER_MARGIN + card_left = margin + card_right = self.width - margin + content_left = card_left + self.padding + content_width = max(card_right - self.padding - content_left, 1) + + brand_font = self._load_font_by_size(max(int(self.font_size * 1.05), 12)) + badge_font = self._load_font_by_size(max(int(self.font_size * 0.8), 11)) + heading_font = self._load_font_by_size(max(int(self.font_size * 1.15), 13)) + footer_font = self._load_font_by_size(max(int(self.font_size * 0.75), 11)) + + header_pad_y = max(int(self.font_size * 0.65), 12) + badge_pad_x = max(int(self.font_size * 0.6), 10) + badge_pad_y = max(int(self.font_size * 0.25), 4) + + badge_text = str(scene_style["label"]) + brand_height = self._text_size(self.BRAND_LABEL, brand_font)[1] + badge_text_height = self._text_size(badge_text, badge_font)[1] + badge_height = badge_text_height + 2 * badge_pad_y + badge_width = int(self._measure_text(badge_text, badge_font)) + 2 * badge_pad_x + header_height = max(brand_height, badge_height) + 2 * header_pad_y + + body_line_spacing = max(self.line_spacing, self.BODY_LINE_SPACING_MIN) + body_line_height = int(self._text_size("测试", self.font)[1] * body_line_spacing) + lines = self._wrap_text(text, content_width) + body_height = len(lines) * body_line_height + + heading_height = 0 + if title: + heading_height = int(self._text_size("测试", heading_font)[1] * 1.5) + int(self.font_size * 0.5) + + footer_line_height = self._text_size(self.FOOTER_TEXT, footer_font)[1] + footer_gap = max(int(self.padding * 0.6), 12) + + header_bottom = margin + header_height + content_top = header_bottom + self.padding + body_top = content_top + heading_height + body_bottom = body_top + body_height + divider_y = body_bottom + self.padding + footer_top = divider_y + footer_gap + card_bottom = footer_top + footer_line_height + self.padding + + height = max(card_bottom + margin, self.MIN_HEIGHT) + + # 绘制卡片与头部 + image = Image.new('RGB', (self.width, height), color=self.OUTER_BG_COLOR) + draw = ImageDraw.Draw(image) + + draw.rounded_rectangle( + [card_left, margin, card_right, height - margin], + radius=self.CARD_RADIUS, + fill=self.bg_color, + ) + draw.rounded_rectangle( + [card_left, margin, card_right, header_bottom], + radius=self.CARD_RADIUS, + fill=self.BRAND_COLOR, + ) + # 抹平蓝色横幅底部圆角，使其与卡片左右边缘齐平 + draw.rectangle( + [card_left, header_bottom - self.CARD_RADIUS, card_right, header_bottom], + fill=self.BRAND_COLOR, + ) + + badge_top = margin + max((header_height - badge_height) // 2, 0) + badge_left = card_right - self.padding - badge_width + draw.rounded_rectangle( + [badge_left, badge_top, badge_left + badge_width, badge_top + badge_height], + radius=badge_height // 2, + fill=self.BADGE_BG_COLOR, + ) + + brand_top = margin + max((header_height - brand_height) // 2, 0) + badge_text_top = badge_top + max((badge_height - badge_text_height) // 2, 0) + + with ExitStack() as stack: + pilmoji: Optional[Pilmoji] = None + if self._emoji_enabled: + try: + pilmoji = stack.enter_context(Pilmoji(image)) + except Exception as exc: + self._emoji_enabled = False + logger.warning("Emoji 渲染器初始化失败，降级为普通文本绘制: %s", exc) + + self._paint(pilmoji, draw, (content_left, brand_top), self.BRAND_LABEL, brand_font, self.BRAND_TEXT_COLOR) + self._paint(pilmoji, draw, (badge_left + badge_pad_x, badge_text_top), badge_text, badge_font, tuple(scene_style["color"])) + + if title: + self._paint(pilmoji, draw, (content_left, content_top), title, heading_font, self.text_color) + + y = body_top + for line in lines: + if line: + self._paint(pilmoji, draw, (content_left, y), line, self.font, self.text_color) + y += body_line_height + + draw.line( + [(content_left, divider_y), (card_right - self.padding, divider_y)], + fill=self.DIVIDER_COLOR, + width=1, + ) + self._paint(pilmoji, draw, (content_left, footer_top), self.FOOTER_TEXT, footer_font, self.FOOTER_TEXT_COLOR) + + # 转换为字节流 + img_byte_arr = io.BytesIO() + image.save(img_byte_arr, format='PNG') + img_byte_arr.seek(0) + + return img_byte_arr.getvalue() + + def render_to_base64(self, text: str, title: Optional[str] = DEFAULT_TITLE, scene: str = "common") -> str: + """ + 将文本渲染为图片并返回 base64 编码 + + Args: + text: 文本内容 + title: 标题文案，默认沿用旧版固定文案；显式传 None 或空字符串时不绘制标题行 + scene: 场景标识，支持 common/group/private，未知值回退为 common + + Returns: + base64 编码的图片数据 + """ + img_bytes = self.render(text, title=title, scene=scene) + base64_str = base64.b64encode(img_bytes).decode('utf-8') + return f"base64://{base64_str}" diff --git a/danding_bot/plugins/danding_qqpush/preview_image_render.py b/danding_bot/plugins/danding_qqpush/preview_image_render.py new file mode 100644 index 0000000..cd18158 --- /dev/null +++ b/danding_bot/plugins/danding_qqpush/preview_image_render.py @@ -0,0 +1,36 @@ +"""预览渲染辅助模块 - 只为超级用户预览命令按插件配置提供渲染器实例。 + +放在独立模块里，避免 test_command 反向依赖尚未完成初始化的插件包 __init__； +渲染逻辑仍然复用 image_render.ImageRenderer，不在此处重复实现。 +""" +from typing import Optional + +from .config import Config +from .image_render import ImageRenderer + +# 渲染器实例缓存：按插件配置构建一次后复用（字体只加载一次） +_preview_renderer: Optional[ImageRenderer] = None + + +def get_preview_renderer(config: Config) -> ImageRenderer: + """ + 获取通知卡片渲染器 + + Args: + config: 插件配置对象 + + Returns: + 缓存的 ImageRenderer 实例 + """ + global _preview_renderer + if _preview_renderer is None: + _preview_renderer = ImageRenderer( + width=config.ImageWidth, + font_size=config.ImageFontSize, + padding=config.ImagePadding, + line_spacing=config.ImageLineSpacing, + bg_color=config.ImageBgColor, + text_color=config.ImageTextColor, + font_paths=config.FontPaths, + ) + return _preview_renderer diff --git a/danding_bot/plugins/danding_qqpush/test_command.py b/danding_bot/plugins/danding_qqpush/test_command.py new file mode 100644 index 0000000..0816b6f --- /dev/null +++ b/danding_bot/plugins/danding_qqpush/test_command.py @@ -0,0 +1,108 @@ +"""通知图片测试命令 - 供超级用户在群聊/私聊中预览统一通知卡片。 + +用法(仅超级用户,群内直接发送或私聊发送,均无需 @机器人): + 测试通知图片 + 通知图片测试 + 测试通知卡片 +""" +import asyncio + +from nonebot import on_command, logger +from nonebot.adapters.onebot.v11 import Bot, Message, MessageEvent, MessageSegment +from nonebot.matcher import Matcher +from nonebot.permission import SUPERUSER + +from .preview_image_render import get_preview_renderer +from .text_parser import TextParser + +# 主命令与别名，唯一的命令入口定义，测试直接据此断言 +TEST_COMMAND = "测试通知图片" +"""主命令""" +TEST_COMMAND_ALIASES = ("通知图片测试", "测试通知卡片") +"""命令别名""" + +# 固定预览内容：覆盖标题、中文、多行、Emoji 与中文标点，便于肉眼核对版式 +PREVIEW_TITLE = "# 会被解析为换行：" +"""预览卡片标题行文案：同时演示 # 作为换行符的用法""" +PREVIEW_TEXT = "\n".join( + ( + "【通知卡片测试】", + "本卡片由「测试通知图片」命令生成，用于核对统一通知卡片版式。", + "中文标点：逗号，句号。顿号、分号；冒号：问号？感叹号！", + "Emoji：🎉 🏇 😀 👍🏻 🇨🇳", + "混合排版：emoji 与正文之间不应错位 🎊，行距与折行都要正常。", + ) +) +"""预览正文（按换行符直接给出，未经 # 替换）""" +PREVIEW_PREVIEW_TEXT = PREVIEW_TEXT.replace("\n", "#") +"""预览文本的 # 写法，经 TextParser 解析后与 PREVIEW_TEXT 一致""" + + +def resolve_scene(event: MessageEvent) -> str: + """ + 按事件来源选择场景徽章 + + Args: + event: 消息事件 + + Returns: + 群聊为 "group"，否则为 "private" + """ + return "group" if getattr(event, "group_id", None) else "private" + + +def build_message(image_base64: str) -> Message: + """ + 构建 OneBot V11 图片消息 + + Args: + image_base64: base64:// 前缀的图片数据 + + Returns: + 可直接发送的 Message + """ + return Message(MessageSegment.image(image_base64)) + + +def preview_text() -> str: + """ + 构造固定预览文本（以 # 表示换行，交由 TextParser 解析） + + Returns: + 预览文本 + """ + return PREVIEW_PREVIEW_TEXT + + +test_notification_image = on_command( + TEST_COMMAND, + aliases=set(TEST_COMMAND_ALIASES), + permission=SUPERUSER, + priority=1, + block=True, +) + + +@test_notification_image.handle() +async def handle_test_notification_image(matcher: Matcher, bot: Bot, event: MessageEvent): + """把固定预览内容渲染成通知卡片并回发到当前会话。""" + from . import plugin_config + + scene = resolve_scene(event) + parsed_text = TextParser(max_length=plugin_config.MaxTextLength).parse(preview_text()) + renderer = get_preview_renderer(plugin_config) + + try: + # Pillow 渲染是同步阻塞操作，放到线程池避免阻塞事件循环 + image_base64 = await asyncio.to_thread( + renderer.render_to_base64, parsed_text, PREVIEW_TITLE, scene + ) + # 用当前事件的 Bot 直接发送 OneBot V11 图片消息，群聊/私聊都回到触发会话 + await bot.send(event, build_message(image_base64)) + logger.info(f"[Danding_QqPush] 通知图片测试成功 - scene={scene}") + except BaseException as e: + # 保留 NoneBot 控制流异常（如 FinishedException），只处理真正的异常 + if not isinstance(e, Exception): + raise + logger.error(f"[Danding_QqPush] 通知图片测试失败 - scene={scene}: {e}") + await matcher.finish("通知图片生成失败，请查看机器人日志后重试。") diff --git a/tests/test_danding_qqpush_image_render.py b/tests/test_danding_qqpush_image_render.py new file mode 100644 index 0000000..4764e93 --- /dev/null +++ b/tests/test_danding_qqpush_image_render.py @@ -0,0 +1,353 @@ +"""通知卡片渲染测试：版式、场景徽章、中文/Emoji 折行与降级路径。""" +import base64 +import importlib.util +import io +from pathlib import Path + +from PIL import Image, ImageDraw + + +def load_image_renderer(): + """按路径加载渲染模块，避免拉起需要 NoneBot 初始化的插件包。""" + path = Path(__file__).resolve().parents[1] / "danding_bot" / "plugins" / "danding_qqpush" / "image_render.py" + spec = importlib.util.spec_from_file_location("danding_qqpush_image_render", path) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def build_renderer(module, **kwargs): + return module.ImageRenderer(**kwargs) + + +def install_offline_pilmoji(monkeypatch, module): + """用不访问远程 Emoji 源的替身，保证 Emoji 相关渲染测试离线可跑。""" + + class OfflinePilmoji: + def __init__(self, image): + self._draw = ImageDraw.Draw(image) + + def __enter__(self): + return self + + def __exit__(self, *args): + return False + + def text(self, xy, text, font=None, fill=None, **kwargs): + self._draw.text(xy, text, font=font, fill=fill) + + monkeypatch.setattr(module, "Pilmoji", OfflinePilmoji) + + +def open_png(data: bytes) -> Image.Image: + return Image.open(io.BytesIO(data)).convert("RGB") + + +def ink_bbox(image: Image.Image, box=None): + """返回区域内墨迹（暗色像素）的包围盒，用于判断是否越界。""" + region = image.crop(box) if box else image + mask = region.convert("L").point(lambda value: 255 if value < 200 else 0) + return mask.getbbox() + + +def ink_pixels(image: Image.Image, box=None) -> int: + """区域内墨迹像素数量，用于比较某块区域是否新增了文字。""" + region = image.crop(box) if box else image + mask = region.convert("L").point(lambda value: 255 if value < 200 else 0) + return mask.histogram()[255] + + +def first_ink_row(image: Image.Image, y_from: int, margin: int) -> int: + """自 y_from 起逐行扫描，返回第一行存在墨迹的 y 坐标。""" + for y in range(y_from, image.size[1] - 1): + if ink_pixels(image, (margin, y, image.size[0] - margin, y + 1)) > 0: + return y + raise AssertionError("区域下方未找到任何墨迹") + + +def header_bottom(image: Image.Image, margin: int, brand_color) -> int: + """沿水平中线向下找到蓝色头部结束的位置。""" + width, height = image.size + pixels = image.load() + for y in range(margin, height): + if pixels[width // 2, y] != brand_color: + return y + raise AssertionError("未找到品牌头部下边界") + + +def color_count(image: Image.Image, box, color) -> int: + colors = image.crop(box).getcolors(maxcolors=image.size[0] * image.size[1]) + return dict((pixel, count) for count, pixel in colors).get(color, 0) + + +def content_limits(renderer, width: int = 800): + margin = renderer.OUTER_MARGIN + card_right = width - margin + return margin + renderer.padding, card_right - renderer.padding + + +def test_render_keeps_legacy_calls_working(): + module = load_image_renderer() + renderer = build_renderer(module) + + png = renderer.render("系统告警") + assert png.startswith(b"\x89PNG") + + encoded = renderer.render_to_base64("系统告警") + assert encoded.startswith("base64://") + assert base64.b64decode(encoded[len("base64://"):]) == png + + # 历史调用方式（位置参数传 title）依然可用 + assert renderer.render("系统告警", "自定义标题").startswith(b"\x89PNG") + assert renderer.render_to_base64("系统告警", title="自定义标题").startswith("base64://") + + +def test_card_uses_cool_gray_backdrop_and_blue_brand_header(): + module = load_image_renderer() + renderer = build_renderer(module) + renderer_class = module.ImageRenderer + + image = open_png(renderer.render("卡片版式校验", scene="group")) + width, height = image.size + margin = renderer_class.OUTER_MARGIN + + assert width == renderer.width == 800 + assert image.getpixel((2, 2)) == renderer_class.OUTER_BG_COLOR + assert image.getpixel((width - 3, height - 3)) == renderer_class.OUTER_BG_COLOR + assert image.getpixel((width // 2, margin + 5)) == renderer_class.BRAND_COLOR + assert image.getpixel((width // 2, height - margin - 5)) == renderer.bg_color + + +def test_scene_badges_distinguish_group_and_private_notifications(): + module = load_image_renderer() + renderer = build_renderer(module) + + common = renderer.render("同一段通知内容", scene="common") + group = renderer.render("同一段通知内容", scene="group") + private = renderer.render("同一段通知内容", scene="private") + + assert len({common, group, private}) == 3 + + margin = renderer.OUTER_MARGIN + header_box = (margin, margin, 800 - margin, margin + 80) + private_image = open_png(private) + common_image = open_png(common) + + # 徽章为白色胶囊，文字使用场景强调色 + assert color_count(private_image, header_box, renderer.BADGE_BG_COLOR) > 200 + assert color_count(private_image, header_box, renderer.SCENE_STYLES["private"]["color"]) > 0 + assert color_count(common_image, header_box, renderer.SCENE_STYLES["common"]["color"]) > 0 + assert color_count(common_image, header_box, renderer.SCENE_STYLES["private"]["color"]) == 0 + + +def test_unknown_scene_falls_back_to_common_style(): + module = load_image_renderer() + renderer = build_renderer(module) + + assert renderer._scene_style("group")["label"] == "群聊通知" + assert renderer._scene_style("private")["label"] == "私聊通知" + + unknown = renderer._scene_style("nope") + assert unknown["label"] == renderer.SCENE_STYLES[renderer.DEFAULT_SCENE]["label"] + assert unknown["color"] == renderer.SCENE_STYLES[renderer.DEFAULT_SCENE]["color"] + + assert renderer.render("未知场景", scene="nope") == renderer.render("未知场景", scene="common") + + +def test_long_text_wraps_without_leaving_the_card(): + module = load_image_renderer() + renderer = build_renderer(module) + + content_left, content_right = content_limits(renderer) + content_width = content_right - content_left + + lines = renderer._wrap_text("超长文本测试，" * 200, content_width) + assert all(renderer._measure_text(line, renderer.font) <= content_width for line in lines) + assert "".join(lines) == "超长文本测试，" * 200 + + image = open_png(renderer.render("超长文本测试，" * 200, scene="group")) + width, height = image.size + margin = renderer.OUTER_MARGIN + header_end = header_bottom(image, margin, renderer.BRAND_COLOR) + body_box = (margin, header_end + 1, width - margin, height - margin) + + left, top, right, bottom = ink_bbox(image, body_box) + absolute = (body_box[0] + left, body_box[1] + top, body_box[0] + right, body_box[1] + bottom) + assert absolute[0] >= content_left + assert absolute[2] <= content_right + assert absolute[3] < height - margin + + +def test_explicit_newlines_and_blank_lines_are_preserved(): + module = load_image_renderer() + renderer = build_renderer(module) + content_left, content_right = content_limits(renderer) + + lines = renderer._wrap_text("第一行\n第二行\n\n第四行", content_right - content_left) + + assert lines == ["第一行", "第二行", "", "第四行"] + + +def test_cjk_punctuation_does_not_start_or_end_a_line(): + module = load_image_renderer() + renderer = build_renderer(module) + content_left, content_right = content_limits(renderer) + content_width = content_right - content_left + + cases = [ + # 28 个全角汉字正好铺满一行，逗号会落在下一行行首 + "甲" * 28 + "，后续内容" + "乙" * 26 + "（括号内容）", + # 27 个汉字加左括号刚好占满，右括号内容会被挤到下一行，左括号不能留在行尾 + "甲" * 27 + "（乙" * 20, + ] + + for text in cases: + lines = renderer._wrap_text(text, content_width) + + assert "".join(lines) == text + for line in lines: + if line: + assert line[0] not in module.ImageRenderer.LINE_START_FORBIDDEN + assert renderer._measure_text(line, renderer.font) <= content_width + for line in lines[:-1]: + assert line[-1] not in module.ImageRenderer.LINE_END_FORBIDDEN + + +def test_emoji_is_measured_as_square_and_kept_whole(): + module = load_image_renderer() + renderer = build_renderer(module) + + atoms = list(renderer._iter_atoms("🏇出发", renderer.font)) + assert atoms[0][0] == "🏇" + assert atoms[0][1] == renderer.font.size + + # ZWJ 组合、肤色修饰、区域指示符组合都不应被拆开 + family, flag = "👨‍👩‍👧", "🇨🇳" + atoms = list(renderer._iter_atoms(f"👍🏻{family}{flag}赞", renderer.font)) + groups = [atom for atom, _ in atoms] + assert "👍🏻" in groups + assert family in groups + assert flag in groups + + +def test_emoji_text_does_not_overflow_the_card(monkeypatch): + module = load_image_renderer() + install_offline_pilmoji(monkeypatch, module) + renderer = build_renderer(module) + + content_left, content_right = content_limits(renderer) + content_width = content_right - content_left + + text = "🏇🎉😀 中文与 Emoji 混排的长通知，" * 20 + lines = renderer._wrap_text(text, content_width) + assert "".join(lines) == text + assert all(renderer._measure_text(line, renderer.font) <= content_width for line in lines) + + image = open_png(renderer.render(text, scene="group")) + width, height = image.size + margin = renderer.OUTER_MARGIN + header_end = header_bottom(image, margin, renderer.BRAND_COLOR) + body_box = (margin, header_end + 1, width - margin, height - margin) + left, _, right, _ = ink_bbox(image, body_box) + assert body_box[0] + left >= content_left + assert body_box[0] + right <= content_right + + +def test_body_line_height_is_at_least_one_point_six(monkeypatch): + module = load_image_renderer() + renderer = build_renderer(module, line_spacing=1.2) + + single = open_png(renderer.render("单行")) + double = open_png(renderer.render("第一行\n第二行")) + line_height = double.size[1] - single.size[1] + + assert line_height > 0 + assert line_height >= int(renderer._text_size("测试", renderer.font)[1] * 1.6) + + +def test_default_title_keeps_legacy_text_and_none_hides_it(monkeypatch): + """默认标题沿用旧版文案（向后兼容），显式 None / 空字符串才隐藏标题行。""" + module = load_image_renderer() + install_offline_pilmoji(monkeypatch, module) + renderer = build_renderer(module) + + assert module.ImageRenderer.DEFAULT_TITLE == "蛋定助手通知您：" + + default_image = open_png(renderer.render("正文内容")) + hidden_image = open_png(renderer.render("正文内容", title=None)) + empty_image = open_png(renderer.render("正文内容", title="")) + custom_image = open_png(renderer.render("正文内容", title="🏇 赛马开赛")) + + # 默认渲染比显式隐藏多一行标题 + assert default_image.size[1] > hidden_image.size[1] + # 显式 None 与空字符串等效，且自定义单行标题与默认标题高度一致 + assert hidden_image.size[1] == empty_image.size[1] + assert custom_image.size[1] == default_image.size[1] + + # 默认渲染在头部下方先画标题行，显式隐藏时同一位置直接是正文 + margin = renderer.OUTER_MARGIN + header_end = header_bottom(default_image, margin, module.ImageRenderer.BRAND_COLOR) + top_band = ( + margin, + header_end + renderer.padding + 1, + default_image.size[0] - margin, + header_end + renderer.padding + 40, + ) + default_band = ink_bbox(default_image, top_band) + hidden_band = ink_bbox(hidden_image, top_band) + + assert default_band is not None and hidden_band is not None + # 隐藏标题时正文紧贴头部内边距开始 + assert first_ink_row(hidden_image, header_end + 1, margin) >= header_end + renderer.padding + # 标题「蛋定助手通知您：」明显宽于正文「正文内容」，据此确认标题行确实绘制了 + assert (default_band[2] - default_band[0]) > (hidden_band[2] - hidden_band[0]) * 1.5 + + # render_to_base64 同样继承默认标题文案 + encoded = renderer.render_to_base64("正文内容") + assert open_png(base64.b64decode(encoded[len("base64://"):])).size == default_image.size + + +def test_empty_text_keeps_placeholder_and_minimum_height(): + module = load_image_renderer() + renderer = build_renderer(module) + + image = open_png(renderer.render("")) + margin = renderer.OUTER_MARGIN + header_end = header_bottom(image, margin, renderer.BRAND_COLOR) + body_box = (margin, header_end + 1, image.size[0] - margin, image.size[1] - margin) + + assert image.size[1] >= module.ImageRenderer.MIN_HEIGHT + assert ink_bbox(image, body_box) is not None + + +def test_render_falls_back_to_plain_text_when_emoji_source_fails(monkeypatch): + module = load_image_renderer() + renderer = build_renderer(module) + + class BrokenPilmoji: + def __init__(self, image): + self.image = image + + def __enter__(self): + return self + + def __exit__(self, *args): + return False + + def text(self, *args, **kwargs): + raise RuntimeError("emoji 源不可用") + + monkeypatch.setattr(module, "Pilmoji", BrokenPilmoji) + + png = renderer.render("降级测试 🏇", scene="private") + + assert png.startswith(b"\x89PNG") + assert renderer._emoji_enabled is False + + +def test_pure_black_text_color_falls_back_to_soft_ink(): + module = load_image_renderer() + + assert build_renderer(module, text_color=(0, 0, 0)).text_color == module.ImageRenderer.DEFAULT_TEXT_COLOR + assert build_renderer(module, text_color=(10, 20, 30)).text_color == (10, 20, 30) + assert module.ImageRenderer.DEFAULT_TEXT_COLOR != (0, 0, 0) diff --git a/tests/test_danding_qqpush_test_command.py b/tests/test_danding_qqpush_test_command.py new file mode 100644 index 0000000..93d47e9 --- /dev/null +++ b/tests/test_danding_qqpush_test_command.py @@ -0,0 +1,538 @@ +"""通知图片测试命令测试：命令/别名、固定预览内容、群聊/私聊场景与发送约定。 + +直接导入插件包会触发 NoneBot 插件初始化，因此这里按路径加载 test_command.py， +同级模块用轻量替身（text_parser 用真实实现，命令匹配与权限用真实 NoneBot 规则）。 +""" +import ast +import asyncio +import base64 +import importlib.util +import io +import logging +import os +import sys +import types +from contextlib import contextmanager +from pathlib import Path + +import nonebot +import pytest +from nonebot.adapters.onebot.v11 import Message, MessageSegment +from nonebot.adapters.onebot.v11.event import MessageEvent +from nonebot.rule import CommandRule + +PLUGIN_DIR = ( + Path(__file__).resolve().parents[1] + / "danding_bot" + / "plugins" + / "danding_qqpush" +).resolve() +PACKAGE_NAME = "danding_bot.plugins.danding_qqpush" + +# 与 .env 中的 SUPERUSERS 一致，用于权限判定 +SUPERUSER_ID = 1424473282 +NON_SUPERUSER_ID = 10001111 + + +class ConfigStub: + """渲染器所需配置字段的轻量替身。""" + + Token = "danding-test" + ImageWidth = 800 + ImageFontSize = 24 + ImagePadding = 30 + ImageLineSpacing = 1.4 + ImageBgColor = (252, 252, 252) + ImageTextColor = (0, 0, 0) + MaxTextLength = 2000 + FontPaths = ["C:/Windows/Fonts/msyh.ttc"] + + +class ImageRendererStub: + """只记录构造参数的替身，用于验证渲染器按配置构建。""" + + def __init__(self, **kwargs): + self.kwargs = kwargs + + def render_to_base64(self, text, title=None, scene="common"): + return "base64://" + base64.b64encode(b"fake-png").decode("utf-8") + + +class FakeMessageEvent: + """仅保留 scene 判定需要的字段。""" + + def __init__(self, group_id=None, user_id=10001): + self.group_id = group_id + self.user_id = user_id + + +class FakeMatcher: + """记录 finish 调用（失败提示走这里）。""" + + def __init__(self): + self.finished = [] + + async def finish(self, message=None, **kwargs): + self.finished.append(message) + + +class RecordingBot: + """记录 bot.send 的 (event, message)。""" + + def __init__(self): + self.sent = [] + + async def send(self, event, message, **kwargs): + self.sent.append((event, message)) + return {"message_id": len(self.sent)} + + +def build_permission_bot(driver): + """构造权限判定用的真实 OneBot V11 Bot。 + + matcher.permission 经 NoneBot 依赖注入会校验 bot 类型， + 必须是真正的 nonebot.adapters.Bot，所以用真实 Adapter + Bot， + 并把已初始化驱动配置中的超级用户列表（字符串化）拷进 Bot 配置， + 保证超级用户判定与真实运行时一致且不产生 pydantic 归一化告警。 + """ + from nonebot.adapters.onebot.v11.adapter import Adapter + from nonebot.adapters.onebot.v11.bot import Bot + + bot = Bot(Adapter(driver), self_id="10000") + bot.config.superusers = {str(user_id) for user_id in driver.config.superusers} + return bot + + +@contextmanager +def silence_nonebot_logs(): + """初始化期间完全静默 NoneBot 日志，避免把 .env 配置打进测试输出。""" + logging.getLogger("nonebot").setLevel(logging.CRITICAL + 1) + try: + yield + finally: + logging.getLogger("nonebot").setLevel(logging.CRITICAL) + + +def ensure_nonebot_initialized(): + """on_command 需要已初始化的 NoneBot：只创建驱动，不启动服务、不加载插件。""" + try: + nonebot.get_driver() + except ValueError: + with silence_nonebot_logs(): + nonebot.init( + log_level="CRITICAL", + driver="~fastapi+~websockets", + host="127.0.0.1", + port=18080, + ) + driver = nonebot.get_driver() + # 保证超级用户判定不受 .env 的 SUPERUSERS 配置影响； + # 与 .env 一致使用字符串形式，避免驱动配置被 int/str 混装归一化 + if str(SUPERUSER_ID) not in driver.config.superusers: + driver.config.superusers.add(str(SUPERUSER_ID)) + return driver + + +def _load_module(name, path): + spec = importlib.util.spec_from_file_location(f"{PACKAGE_NAME}.{name}", path) + module = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = module + spec.loader.exec_module(module) + return module + + +def load_test_command(): + """按路径加载 test_command.py，同时用替身满足同级模块依赖。""" + if PACKAGE_NAME in sys.modules: + return sys.modules[PACKAGE_NAME + ".test_command"] + + ensure_nonebot_initialized() + + package = types.ModuleType(PACKAGE_NAME) + package.__path__ = [str(PLUGIN_DIR)] + sys.modules[PACKAGE_NAME] = package + + # text_parser 用真实实现：预览文本要经它把 # 解析为换行 + _load_module("text_parser", PLUGIN_DIR / "text_parser.py") + + config_module = types.ModuleType(PACKAGE_NAME + ".config") + config_module.Config = ConfigStub + sys.modules[config_module.__name__] = config_module + + image_render_module = types.ModuleType(PACKAGE_NAME + ".image_render") + image_render_module.ImageRenderer = ImageRendererStub + sys.modules[image_render_module.__name__] = image_render_module + + # preview_image_render 是 test_command 唯一依赖的同级辅助模块 + _load_module("preview_image_render", PLUGIN_DIR / "preview_image_render.py") + + module = _load_module("test_command", PLUGIN_DIR / "test_command.py") + # 供 handler 测试使用的插件配置替身 + package.plugin_config = ConfigStub() + return module + + +def load_real_image_renderer(): + """按路径加载真实渲染模块（不引入 NoneBot）。""" + name = PACKAGE_NAME + ".real_image_render" + if name in sys.modules: + return sys.modules[name] + # 真实 image_render 不含相对导入，直接按路径加载即可 + path = PLUGIN_DIR / "image_render.py" + spec = importlib.util.spec_from_file_location(name, path) + module = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = module + spec.loader.exec_module(module) + return module + + +def source_tree(): + tree = ast.parse((PLUGIN_DIR / "test_command.py").read_text(encoding="utf-8")) + if os.getenv("DSH_DEBUG_QQPUSH"): + print( + "\nDEBUG async nodes:", + [ + (n.name, [is_handle_decorator(d) for d in n.decorator_list]) + for n in ast.walk(tree) + if isinstance(n, ast.AsyncFunctionDef) + ], + ) + return tree + + +def source_text() -> str: + return (PLUGIN_DIR / "test_command.py").read_text(encoding="utf-8") + + +def build_event(text: str, group_id: int | None = None, user_id: int = SUPERUSER_ID): + """构造 OneBot V11 消息事件，用于命令匹配与权限判定。""" + if group_id is None: + return MessageEvent( + time=0, + self_id=10000, + post_type="message", + sub_type="friend", + user_id=user_id, + message_type="private", + message_id=1, + message=Message(text), + original_message=Message(text), + raw_message=text, + font=0, + sender={"user_id": user_id}, + ) + return MessageEvent( + time=0, + self_id=10000, + post_type="message", + sub_type="normal", + user_id=user_id, + message_type="group", + message_id=1, + message=Message(text), + original_message=Message(text), + raw_message=text, + font=0, + sender={"user_id": user_id}, + group_id=group_id, + ) + + +def command_matches(matcher, event) -> bool: + """按 NoneBot 依赖注入的语义执行命令匹配规则。""" + state: dict = {} + for checker in matcher.rule.checkers: + if isinstance(checker.call, CommandRule): + state = { + "_prefix": { + "command": (event.get_plaintext().strip(),), + "command_arg": None, + "command_whitespace": None, + } + } + break + return asyncio.run(matcher.rule(build_permission_bot(nonebot.get_driver()), event, state)) + + +def has_permission(matcher, event) -> bool: + """按 NoneBot 运行时约定调用 matcher 的 permission（Superuser 需要 bot.adapter/bot.config）。""" + return asyncio.run( + matcher.permission(build_permission_bot(nonebot.get_driver()), event) + ) + + +def test_main_command_and_aliases_are_registered_and_match_plain_messages(): + """主命令「测试通知图片」与两个别名都要能直接触发（群聊、私聊都不需要 @机器人）。""" + module = load_test_command() + matcher = module.test_notification_image + + assert module.TEST_COMMAND == "测试通知图片" + assert set(module.TEST_COMMAND_ALIASES) == {"通知图片测试", "测试通知卡片"} + + for text in (module.TEST_COMMAND, *module.TEST_COMMAND_ALIASES): + for group_id in (None, 621016172): + assert command_matches( + matcher, build_event(text, group_id) + ), f"命令 {text} 应可触发（group_id={group_id}）" + + # 无关消息不触发 + for text in ("测试通知", "通知图片", "测试通知图片额外内容"): + assert not command_matches(matcher, build_event(text)) + + +def test_command_is_superuser_only_and_not_at_restricted(): + module = load_test_command() + matcher = module.test_notification_image + + call = command_registration_source() + keywords = keyword_names(call) + + permission = keywords["permission"] + assert isinstance(permission, ast.Name) and permission.id == "SUPERUSER" + assert any( + isinstance(node, ast.ImportFrom) + and node.module == "nonebot.permission" + and any(alias.name == "SUPERUSER" for alias in node.names) + for node in ast.walk(source_tree()) + ) + + # 群聊与私聊都要能触发：不限制 to_me，也不限制会话类型 + assert "rule" not in keywords + assert ast.literal_eval(keywords["block"]) is True + assert "to_me" not in source_text() + + # 权限判定走真实 NoneBot Permission 对象 + assert has_permission(matcher, build_event(module.TEST_COMMAND)) is True + assert not has_permission( + matcher, build_event(module.TEST_COMMAND, user_id=NON_SUPERUSER_ID) + ) + + +def test_single_handler_is_registered_on_the_matcher(): + handlers = [ + node + for node in ast.walk(source_tree()) + if isinstance(node, ast.AsyncFunctionDef) + if any(is_handle_decorator(decorator) for decorator in node.decorator_list) + ] + assert [node.name for node in handlers] == ["handle_test_notification_image"] + assert [arg.arg for arg in handlers[0].args.args] == ["matcher", "bot", "event"] + # 使用当前事件的 Bot 发送，渲染放到线程池 + assert "bot.send(event, build_message(image_base64))" in source_text() + assert "asyncio.to_thread" in source_text() + + +def test_plugin_package_registers_the_test_command(): + """插件包导入时须加载该模块，命令才会真正注册。""" + source = (PLUGIN_DIR / "__init__.py").read_text(encoding="utf-8") + + assert any( + isinstance(node, ast.ImportFrom) + and node.module is None + and any(alias.name == "test_command" for alias in node.names) + for node in ast.walk(ast.parse(source)) + ), "插件 __init__.py 应以相对导入加载 test_command" + assert "测试通知图片" in source + assert source.index("plugin_config = Config.model_validate") < source.index( + "from . import test_command" + ), "插件包须先完成配置初始化，再导入测试命令" + + +def test_preview_content_covers_title_chinese_multiline_emoji_and_punctuation(): + module = load_test_command() + text = module.preview_text() + + assert module.PREVIEW_TITLE + # 标题同时演示 # 作为换行符的用法 + assert module.PREVIEW_TITLE != module.PREVIEW_TITLE.replace("#", "") + + # 标题、中文、多行、Emoji 与中文标点都要有，方便肉眼核对版式 + assert "#" in text, "预览文本应使用 # 表示换行" + assert any("\u4e00" <= char <= "\u9fff" for char in text), "预览应包含中文" + assert len(text.split("#")) >= 3, "预览应为多行内容" + assert any(char in text for char in "，。、；：！？"), "预览应包含中文标点" + assert any(ord(char) > 0x1F000 for char in text), "预览应包含 Emoji" + + # 经 TextParser 解析后，与 PREVIEW_TEXT 完全一致（# 均表示换行） + parser = sys.modules[PACKAGE_NAME + ".text_parser"].TextParser( + max_length=ConfigStub.MaxTextLength + ) + assert parser.parse(text) == module.PREVIEW_TEXT + assert module.PREVIEW_TEXT.count("\n") == text.count("#") + assert "#" not in module.PREVIEW_TEXT + + +def test_scene_resolution_uses_group_for_group_events_and_private_otherwise(): + module = load_test_command() + + assert module.resolve_scene(FakeMessageEvent(group_id=123456789)) == "group" + assert module.resolve_scene(FakeMessageEvent()) == "private" + # group_id=0 视为无效群号，同样回退到私聊 + assert module.resolve_scene(FakeMessageEvent(group_id=0)) == "private" + assert module.resolve_scene(build_event("测试通知图片", 621016172)) == "group" + assert module.resolve_scene(build_event("测试通知图片")) == "private" + + # 两个场景徽章都由现有渲染器提供，测试命令不新增场景 + real = load_real_image_renderer() + assert "group" in real.ImageRenderer.SCENE_STYLES + assert "private" in real.ImageRenderer.SCENE_STYLES + + +def test_render_runs_in_thread_and_passes_scene_and_title(monkeypatch): + module = load_test_command() + calls = [] + + class RecordingRenderer: + def render_to_base64(self, text, title=None, scene="common"): + calls.append({"text": text, "title": title, "scene": scene}) + return "base64://cG5n" + + monkeypatch.setattr(module, "get_preview_renderer", lambda config: RecordingRenderer()) + real_to_thread = asyncio.to_thread + to_thread_calls = [] + + async def spy_to_thread(func, *args, **kwargs): + to_thread_calls.append(func) + return await real_to_thread(func, *args, **kwargs) + + monkeypatch.setattr(module.asyncio, "to_thread", spy_to_thread) + + for group_id, expected_scene in ((987654321, "group"), (None, "private")): + calls.clear() + to_thread_calls.clear() + matcher = FakeMatcher() + bot = RecordingBot() + + asyncio.run( + module.handle_test_notification_image( + matcher, bot, FakeMessageEvent(group_id=group_id) + ) + ) + + assert to_thread_calls, "Pillow 渲染必须放到 asyncio.to_thread" + assert [call["scene"] for call in calls] == [expected_scene] + assert calls[0]["title"] == module.PREVIEW_TITLE + assert calls[0]["text"] == module.PREVIEW_TEXT + assert calls[0]["text"] == module.preview_text().replace("#", "\n") + assert not matcher.finished + + +def test_handler_sends_onebot_v11_image_message_via_event_bot(): + module = load_test_command() + matcher = FakeMatcher() + bot = RecordingBot() + event = FakeMessageEvent(group_id=1) + + rendered = "base64://" + base64.b64encode(b"png-bytes").decode("utf-8") + message = module.build_message(rendered) + + assert message[0].type == "image" + assert message[0].data["file"] == rendered + assert module.build_message(rendered) == message + + asyncio.run(module.handle_test_notification_image(matcher, bot, event)) + + assert len(bot.sent) == 1 + sent_event, sent_message = bot.sent[0] + assert sent_event is event + assert sent_message[0].type == "image" + assert sent_message[0].data["file"].startswith("base64://") + assert not matcher.finished + + +def test_render_failure_logs_and_replies_with_readable_hint(monkeypatch): + module = load_test_command() + matcher = FakeMatcher() + bot = RecordingBot() + logged = [] + + class BrokenRenderer: + def render_to_base64(self, text, title=None, scene="common"): + raise RuntimeError("字体缺失") + + monkeypatch.setattr(module, "get_preview_renderer", lambda config: BrokenRenderer()) + monkeypatch.setattr(module.logger, "error", lambda message: logged.append(str(message))) + + asyncio.run(module.handle_test_notification_image(matcher, bot, FakeMessageEvent())) + + assert not bot.sent + assert len(matcher.finished) == 1 + assert isinstance(matcher.finished[0], str) + assert "字体缺失" in logged[0] + assert "scene=private" in logged[0] + + +def test_preview_renderer_is_built_from_plugin_config_and_reused(): + module = load_test_command() + helper = sys.modules[PACKAGE_NAME + ".preview_image_render"] + helper._preview_renderer = None + + first = module.get_preview_renderer(ConfigStub()) + second = module.get_preview_renderer(ConfigStub()) + + assert isinstance(first, ImageRendererStub) + assert first is second, "渲染器实例应被缓存复用" + assert first.kwargs["width"] == ConfigStub.ImageWidth + assert first.kwargs["font_size"] == ConfigStub.ImageFontSize + assert first.kwargs["padding"] == ConfigStub.ImagePadding + assert first.kwargs["line_spacing"] == ConfigStub.ImageLineSpacing + assert tuple(first.kwargs["bg_color"]) == tuple(ConfigStub.ImageBgColor) + assert tuple(first.kwargs["text_color"]) == tuple(ConfigStub.ImageTextColor) + assert list(first.kwargs["font_paths"]) == list(ConfigStub.FontPaths) + + helper._preview_renderer = None + + +def test_preview_renders_through_the_shared_renderer(): + """预览内容必须能真正渲染成通知卡片（使用真实 ImageRenderer）。""" + module = load_test_command() + real = load_real_image_renderer() + renderer = real.ImageRenderer( + width=800, font_size=24, padding=30, font_paths=ConfigStub.FontPaths + ) + + png = renderer.render(module.PREVIEW_TEXT, title=module.PREVIEW_TITLE, scene="group") + assert png.startswith(b"\x89PNG") + + from PIL import Image + + image = Image.open(io.BytesIO(png)) + assert image.size[0] == 800 + assert image.size[1] >= real.ImageRenderer.MIN_HEIGHT + + encoded = renderer.render_to_base64( + module.PREVIEW_TEXT, module.PREVIEW_TITLE, "private" + ) + assert encoded.startswith("base64://") + assert base64.b64decode(encoded[len("base64://"):]) == renderer.render( + module.PREVIEW_TEXT, title=module.PREVIEW_TITLE, scene="private" + ) + + +def is_handle_decorator(decorator) -> bool: + """识别 @matcher.handle 与 @matcher.handle() 两种等价写法。""" + target = decorator.func if isinstance(decorator, ast.Call) else decorator + return isinstance(target, ast.Attribute) and target.attr == "handle" + + +def command_registration_source(): + """解析 test_command.py 中 on_command 的注册调用。""" + for node in ast.walk(source_tree()): + if not isinstance(node, ast.Call): + continue + func = node.func + if isinstance(func, ast.Name) and func.id == "on_command": + return node + raise AssertionError("test_command.py 中未找到 on_command 注册调用") + + +def keyword_names(call): + return { + keyword.arg: keyword.value + for keyword in call.keywords + if keyword.arg is not None + } + + +if __name__ == "__main__": # pragma: no cover + raise SystemExit(pytest.main([__file__, "-q"]))