xia.silei/DanDingNoneBot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 新增通知图片测试命令

Browse Source
This commit is contained in:
夏斯雷
2026-09-22 21:50:27 +08:00
parent 87c59a0335
commit f4d3abcd09
9 changed files with 2068 additions and 680 deletions
Download Patch File Download Diff File Expand all files Collapse all files

83
danding_bot/plugins/danding_qqpush/README.md
View File

@@ -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
```

137
danding_bot/plugins/danding_qqpush/__init__.py
View File

@@ -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}")

460
danding_bot/plugins/danding_qqpush/api.py
View File

@@ -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

688
danding_bot/plugins/danding_qqpush/image_render.py
View File

@@ -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}"

36
danding_bot/plugins/danding_qqpush/preview_image_render.py Normal file
View File

@@ -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

108
danding_bot/plugins/danding_qqpush/test_command.py Normal file
View File

@@ -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("通知图片生成失败，请查看机器人日志后重试。")