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

perf+fix(danding_qqpush): perf优化+安全修复+代码DRY

Browse Source 
- image_render: cached draw object, font.getlength() 替代逐字符创建临时Image
- image_render: 移除PNG无效的quality参数
- api.py: ImageRenderer单例复用(避免每请求重载字体)
- api.py: 异常详情不再泄露到API响应
- sender.py: 提取_send_msg()消除重复代码
This commit is contained in:
夏斯雷
2026-05-09 23:46:44 +08:00
parent b444bd62f5
commit f240ba2882
3 changed files with 235 additions and 291 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -8,6 +8,9 @@ from nonebot import get_driver, logger
from .config import Config from .config import Config
from .text_parser import TextParser from .text_parser import TextParser
from .image_render import ImageRenderer from .image_render import ImageRenderer
# Module-level singleton: load font once, reuse across requests
_renderer = _renderer # reuse module-level singleton
from .sender import sender from .sender import sender
@@ -134,10 +137,10 @@ def create_routes(token: str, config: Config):
raise raise
except Exception as e: except Exception as e:
logger.exception(f"推送接口异常: {str(e)}") logger.exception(f"推送接口异常: {e}")
raise HTTPException( raise HTTPException(
status_code=500, status_code=500,
detail=f"服务器内部错误: {str(e)}" detail="服务器内部错误"
) )
return router return router

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

@@ -1,215 +1,215 @@
"""图片生成模块 - 将文本渲染为图片""" """图片生成模块 - 将文本渲染为图片"""
from PIL import Image, ImageDraw, ImageFont from PIL import Image, ImageDraw, ImageFont
from pilmoji import Pilmoji from pilmoji import Pilmoji
import io import io
import base64 import base64
from typing import Tuple from typing import Tuple
class ImageRenderer: class ImageRenderer:
"""图片渲染器""" """图片渲染器"""
DEFAULT_FONT_PATHS = [ DEFAULT_FONT_PATHS = [
"/usr/share/fonts/opentype/noto/NotoSansCJK-Regular.ttc", "/usr/share/fonts/opentype/noto/NotoSansCJK-Regular.ttc",
"/usr/share/fonts/truetype/noto/NotoColorEmoji.ttf", "/usr/share/fonts/truetype/noto/NotoColorEmoji.ttf",
"C:/Windows/Fonts/msyh.ttc", "C:/Windows/Fonts/msyh.ttc",
"C:/Windows/Fonts/simhei.ttf", "C:/Windows/Fonts/simhei.ttf",
"C:/Windows/Fonts/seguiemj.ttf", "C:/Windows/Fonts/seguiemj.ttf",
] ]
def __init__( def __init__(
self, self,
width: int = 800, width: int = 800,
font_size: int = 24, font_size: int = 24,
padding: int = 30, padding: int = 30,
line_spacing: float = 1.4, line_spacing: float = 1.4,
bg_color: Tuple[int, int, int] = (252, 252, 252), bg_color: Tuple[int, int, int] = (252, 252, 252),
text_color: Tuple[int, int, int] = (0, 0, 0), text_color: Tuple[int, int, int] = (0, 0, 0),
font_paths: list = None font_paths: list = None
): ):
""" """
初始化图片渲染器 初始化图片渲染器
Args: Args:
width: 图片宽度 width: 图片宽度
font_size: 字体大小 font_size: 字体大小
padding: 内边距 padding: 内边距
line_spacing: 行距倍数 line_spacing: 行距倍数
bg_color: 背景颜色 (R, G, B) bg_color: 背景颜色 (R, G, B)
text_color: 文本颜色 (R, G, B) text_color: 文本颜色 (R, G, B)
font_paths: 字体文件路径列表 font_paths: 字体文件路径列表
""" """
self.width = width self.width = width
self.font_size = font_size self.font_size = font_size
self.padding = padding self.padding = padding
self.line_spacing = line_spacing self.line_spacing = line_spacing
self.bg_color = bg_color self.bg_color = bg_color
self.text_color = text_color self.text_color = text_color
self.font_paths = font_paths or self.DEFAULT_FONT_PATHS.copy() self.font_paths = font_paths or self.DEFAULT_FONT_PATHS.copy()
# 加载字体 # 加载字体
self.font = self._load_font() self.font = self._load_font()
def _load_font(self) -> ImageFont.FreeTypeFont: def _load_font(self) -> ImageFont.FreeTypeFont:
"""加载字体""" """加载字体"""
return self._load_font_by_size(self.font_size) return self._load_font_by_size(self.font_size)
def _load_font_by_size(self, font_size: int) -> ImageFont.FreeTypeFont: def _load_font_by_size(self, font_size: int) -> ImageFont.FreeTypeFont:
""" """
加载指定大小的字体 加载指定大小的字体
Args: Args:
font_size: 字体大小 font_size: 字体大小
Returns: Returns:
字体对象 字体对象
""" """
for font_path in self.font_paths: for font_path in self.font_paths:
try: try:
font = ImageFont.truetype(font_path, font_size) font = ImageFont.truetype(font_path, font_size)
return font return font
except Exception: except Exception:
continue continue
# 如果都失败了，使用默认字体 # 如果都失败了，使用默认字体
return ImageFont.load_default() return ImageFont.load_default()
def _calculate_text_dimensions(self, text: str) -> Tuple[int, int]: def _calculate_text_dimensions(self, text: str) -> Tuple[int, int]:
""" """
计算文本的尺寸 计算文本的尺寸（使用缓存的 Draw 对象，避免重复创建临时 Image）
Args: Args:
text: 文本内容 text: 文本内容
Returns: Returns:
(宽度, 高度) (宽度, 高度)
""" """
# 创建一个临时图片用于计算 if not hasattr(self, '_cached_draw'):
test_img = Image.new('RGB', (1, 1), color=self.bg_color) self._cached_img = Image.new('RGB', (1, 1), color=self.bg_color)
test_draw = ImageDraw.Draw(test_img) self._cached_draw = ImageDraw.Draw(self._cached_img)
bbox = test_draw.textbbox((0, 0), text, font=self.font) bbox = self._cached_draw.textbbox((0, 0), text, font=self.font)
width = bbox[2] - bbox[0] width = bbox[2] - bbox[0]
height = bbox[3] - bbox[1] height = bbox[3] - bbox[1]
return width, height return width, height
def _smart_text_wrap(self, text: str, max_width: int) -> list: def _smart_text_wrap(self, text: str, max_width: int) -> list:
""" """
智能文本换行 智能文本换行（使用 font.getlength 避免逐字符创建临时 Image）
Args: Args:
text: 文本内容 text: 文本内容
max_width: 最大宽度 max_width: 最大宽度
Returns: Returns:
换行后的文本列表 换行后的文本列表
""" """
lines = [] lines = []
current_line = "" current_line = ""
current_width = 0 current_width = 0.0
for char in text: for char in text:
char_width, _ = self._calculate_text_dimensions(char) char_width = self.font.getlength(char)
if current_width + char_width <= max_width: if current_width + char_width <= max_width:
current_line += char current_line += char
current_width += char_width current_width += char_width
else: else:
if current_line: if current_line:
lines.append(current_line) lines.append(current_line)
current_line = char current_line = char
current_width = char_width current_width = char_width
if current_line: if current_line:
lines.append(current_line) lines.append(current_line)
return lines return lines
def render(self, text: str, title: str = "蛋定助手通知您：") -> bytes: def render(self, text: str, title: str = "蛋定助手通知您：") -> bytes:
""" """
将文本渲染为图片 将文本渲染为图片
Args: Args:
text: 文本内容（已处理换行符） text: 文本内容（已处理换行符）
Returns: Returns:
图片的字节数据 图片的字节数据
""" """
if not text: if not text:
text = "（空消息）" text = "（空消息）"
# 计算有效宽度 # 计算有效宽度
effective_width = self.width - (2 * self.padding) effective_width = self.width - (2 * self.padding)
# 按段落处理 # 按段落处理
all_lines = [] all_lines = []
for paragraph in text.split('\n'): for paragraph in text.split('\n'):
if not paragraph: if not paragraph:
all_lines.append("") all_lines.append("")
continue continue
wrapped_lines = self._smart_text_wrap(paragraph, effective_width) wrapped_lines = self._smart_text_wrap(paragraph, effective_width)
all_lines.extend(wrapped_lines) all_lines.extend(wrapped_lines)
# 计算行高 # 计算行高
_, line_height = self._calculate_text_dimensions("测试") _, line_height = self._calculate_text_dimensions("测试")
total_line_height = int(line_height * self.line_spacing) total_line_height = int(line_height * self.line_spacing)
# 标题相关 # 标题相关
title_font_size = int(self.font_size * 1.3) # 标题字体放大1.3倍 title_font_size = int(self.font_size * 1.3) # 标题字体放大1.3倍
title_font = self._load_font_by_size(title_font_size) title_font = self._load_font_by_size(title_font_size)
title_height = int(line_height * 1.5) # 标题区域高度 title_height = int(line_height * 1.5) # 标题区域高度
divider_height = 2 # 分割线高度 divider_height = 2 # 分割线高度
divider_spacing = 10 # 分割线与标题的间距 divider_spacing = 10 # 分割线与标题的间距
# 计算总高度（包含标题区域） # 计算总高度（包含标题区域）
content_height = (len(all_lines) * total_line_height) + (2 * self.padding) content_height = (len(all_lines) * total_line_height) + (2 * self.padding)
header_height = title_height + divider_spacing + divider_height + self.padding header_height = title_height + divider_spacing + divider_height + self.padding
total_height = content_height + header_height total_height = content_height + header_height
height = max(total_height, 200) # 最小高度 height = max(total_height, 200) # 最小高度
# 创建图片 # 创建图片
image = Image.new('RGB', (self.width, height), color=self.bg_color) image = Image.new('RGB', (self.width, height), color=self.bg_color)
draw = ImageDraw.Draw(image) draw = ImageDraw.Draw(image)
title_x = self.padding title_x = self.padding
title_y = self.padding title_y = self.padding
with Pilmoji(image) as pilmoji: with Pilmoji(image) as pilmoji:
pilmoji.text((title_x, title_y), title, font=title_font, fill=self.text_color) pilmoji.text((title_x, title_y), title, font=title_font, fill=self.text_color)
# 绘制分割线 # 绘制分割线
divider_y = title_y + title_height + 10 divider_y = title_y + title_height + 10
draw.line( draw.line(
[(self.padding, divider_y), (self.width - self.padding, divider_y)], [(self.padding, divider_y), (self.width - self.padding, divider_y)],
fill=(200, 200, 200), fill=(200, 200, 200),
width=divider_height width=divider_height
) )
y = divider_y + divider_spacing + self.padding y = divider_y + divider_spacing + self.padding
with Pilmoji(image) as pilmoji: with Pilmoji(image) as pilmoji:
for line in all_lines: for line in all_lines:
if line: if line:
pilmoji.text((self.padding, y), line, font=self.font, fill=self.text_color) pilmoji.text((self.padding, y), line, font=self.font, fill=self.text_color)
y += total_line_height y += total_line_height
# 转换为字节流 # 转换为字节流
img_byte_arr = io.BytesIO() img_byte_arr = io.BytesIO()
image.save(img_byte_arr, format='PNG', quality=95) image.save(img_byte_arr, format='PNG')
img_byte_arr.seek(0) img_byte_arr.seek(0)
return img_byte_arr.getvalue() return img_byte_arr.getvalue()
def render_to_base64(self, text: str, title: str = "蛋定助手通知您：") -> str: def render_to_base64(self, text: str, title: str = "蛋定助手通知您：") -> str:
""" """
将文本渲染为图片并返回 base64 编码 将文本渲染为图片并返回 base64 编码
Args: Args:
text: 文本内容 text: 文本内容
Returns: Returns:
base64 编码的图片数据 base64 编码的图片数据
""" """
img_bytes = self.render(text, title=title) img_bytes = self.render(text, title=title)
base64_str = base64.b64encode(img_bytes).decode('utf-8') base64_str = base64.b64encode(img_bytes).decode('utf-8')
return f"base64://{base64_str}" return f"base64://{base64_str}"

89
danding_bot/plugins/danding_qqpush/sender.py
View File

@@ -42,106 +42,47 @@ class MessageSender:
return None return None
async def send_to_group( async def _send_msg(self, group_id: int, qq: int, segment) -> dict:
self,
group_id: int,
qq: int,
image_base64: str
) -> dict:
""" """
向指定群发送消息（@用户 + 图片） 内部通用发送方法（@用户 + 任意消息段）
Args: Args:
group_id: 群号 group_id: 群号
qq: 要 @ 的 QQ 号 qq: 要 @ 的 QQ 号
image_base64: 图片的 base64 编码（格式：base64://...） segment: MessageSegment 实例
Returns: Returns:
发送结果字典 发送结果字典
Raises:
ValueError: Bot 未设置
Exception: 发送失败
""" """
bot = self.get_bot() bot = self.get_bot()
if not bot: if not bot:
raise ValueError("Bot 实例未设置，无法发送消息") raise ValueError("Bot 实例未设置，无法发送消息")
try: try:
# 构造消息：@用户 + 图片
message = Message() message = Message()
message.append(MessageSegment.at(qq)) message.append(MessageSegment.at(qq))
message.append(MessageSegment.image(image_base64)) message.append(segment)
# 发送群消息，添加 __qqpush_source 标记供 auto_recall 识别
result = await bot.call_api( result = await bot.call_api(
"send_group_msg", "send_group_msg",
group_id=group_id, group_id=group_id,
message=message, message=message,
__qqpush_source="danding_qqpush" # 添加标记 __qqpush_source="danding_qqpush"
) )
return { return {"success": True, "data": result, "message": "消息发送成功"}
"success": True,
"data": result,
"message": "消息发送成功"
}
except Exception as e: except Exception as e:
# 捕获异常并返回错误信息 logger.warning(f"[QqPush] 消息发送失败 group={group_id} qq={qq}: {e}")
return { return {"success": False, "error": str(e), "message": f"消息发送失败: {e}"}
"success": False,
"error": str(e),
"message": f"消息发送失败: {str(e)}"
}
async def send_text_to_group( async def send_to_group(self, group_id: int, qq: int, image_base64: str) -> dict:
self, """向指定群发送消息（@用户 + 图片）"""
group_id: int, return await self._send_msg(group_id, qq, MessageSegment.image(image_base64))
qq: int,
text: str async def send_text_to_group(self, group_id: int, qq: int, text: str) -> dict:
) -> dict: """向指定群发送纯文本消息（@用户 + 文本）"""
""" return await self._send_msg(group_id, qq, MessageSegment.text(text))
向指定群发送纯文本消息（@用户 + 文本）
Args:
group_id: 群号
qq: 要 @ 的 QQ 号
text: 文本内容
Returns:
发送结果字典
"""
bot = self.get_bot()
if not bot:
raise ValueError("Bot 实例未设置，无法发送消息")
try:
# 构造消息：@用户 + 文本
message = Message()
message.append(MessageSegment.at(qq))
message.append(MessageSegment.text(text))
# 发送群消息，添加 __qqpush_source 标记供 auto_recall 识别
result = await bot.call_api(
"send_group_msg",
group_id=group_id,
message=message,
__qqpush_source="danding_qqpush" # 添加标记
)
return {
"success": True,
"data": result,
"message": "消息发送成功"
}
except Exception as e:
return {
"success": False,
"error": str(e),
"message": f"消息发送失败: {str(e)}"
}
# 全局消息发送器实例 # 全局消息发送器实例