feat(group_horse_racing): 增加赛马消息图片渲染功能

- 在配置中新增图片渲染相关参数:RACE_RENDER_AS_IMAGE、RACE_IMAGE_WIDTH 等
- 复用 danding_qqpush 的 ImageRenderer,使其支持自定义标题
- 在比赛开始、结束和进度播报时,将文本消息转换为带标题的图片发送
- 修复测试用例中的消息发送函数调用
This commit is contained in:
2026-04-04 22:09:53 +08:00
parent 8adc17d311
commit ab1329042a
4 changed files with 49 additions and 7 deletions

View File

@@ -117,7 +117,7 @@ class ImageRenderer:
return lines return lines
def render(self, text: str) -> bytes: def render(self, text: str, title: str = "蛋定助手通知您:") -> bytes:
""" """
将文本渲染为图片 将文本渲染为图片
@@ -148,7 +148,6 @@ class ImageRenderer:
total_line_height = int(line_height * self.line_spacing) total_line_height = int(line_height * self.line_spacing)
# 标题相关 # 标题相关
title = "蛋定助手通知您:"
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) # 标题区域高度
@@ -192,7 +191,7 @@ class ImageRenderer:
return img_byte_arr.getvalue() return img_byte_arr.getvalue()
def render_to_base64(self, text: str) -> str: def render_to_base64(self, text: str, title: str = "蛋定助手通知您:") -> str:
""" """
将文本渲染为图片并返回 base64 编码 将文本渲染为图片并返回 base64 编码
@@ -202,6 +201,6 @@ class ImageRenderer:
Returns: Returns:
base64 编码的图片数据 base64 编码的图片数据
""" """
img_bytes = self.render(text) 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}"

View File

@@ -3,8 +3,10 @@ import uuid
from datetime import datetime from datetime import datetime
from nonebot import on_command from nonebot import on_command
from nonebot.adapters.onebot.v11 import Bot, Event, GroupMessageEvent, PrivateMessageEvent from nonebot.adapters.onebot.v11 import Bot, Event, GroupMessageEvent, Message, MessageSegment, PrivateMessageEvent
from danding_bot.plugins.danding_qqpush.config import Config as QqPushConfig
from danding_bot.plugins.danding_qqpush.image_render import ImageRenderer
from .room_store import RoomStore from .room_store import RoomStore
from .points_service import PointsService from .points_service import PointsService
from .race_engine import RaceEngine from .race_engine import RaceEngine
@@ -18,6 +20,39 @@ room_store = RoomStore(config)
points_service = PointsService(config) points_service = PointsService(config)
race_engine = RaceEngine(config) race_engine = RaceEngine(config)
message_service = MessageService(config) message_service = MessageService(config)
_race_image_renderer: ImageRenderer | None = None
def _get_race_image_renderer() -> ImageRenderer:
global _race_image_renderer
if _race_image_renderer is None:
qqpush_config = QqPushConfig()
_race_image_renderer = ImageRenderer(
width=config.RACE_IMAGE_WIDTH,
font_size=config.RACE_IMAGE_FONT_SIZE,
padding=config.RACE_IMAGE_PADDING,
line_spacing=config.RACE_IMAGE_LINE_SPACING,
font_paths=qqpush_config.FontPaths,
)
return _race_image_renderer
def _build_race_image_message(message: str) -> Message:
if message.startswith("比赛开始!"):
title = "🏇 赛马开赛"
body = message.replace("比赛开始!", "发令枪响,比赛正式开始!", 1)
elif message.startswith("比赛结束!"):
title = "🏆 赛马结果"
body = message
else:
title = "📣 赛马进度"
body = f"🏁 实时播报\n{message}"
renderer = _get_race_image_renderer()
image_base64 = renderer.render_to_base64(body, title=title)
message_obj = Message()
message_obj.append(MessageSegment.image(image_base64))
return message_obj
def get_scope(event: Event) -> str: def get_scope(event: Event) -> str:
@@ -100,11 +135,14 @@ async def settle_race(room: Room):
async def _send_to_scope(bot: Bot, scope: str, message: str): async def _send_to_scope(bot: Bot, scope: str, message: str):
"""Send message to group or private chat based on scope.""" """Send message to group or private chat based on scope."""
try: try:
outbound_message: str | Message = message
if config.RACE_RENDER_AS_IMAGE:
outbound_message = _build_race_image_message(message)
await bot.send_msg( await bot.send_msg(
message_type="group" if scope.startswith("group_") else "private", message_type="group" if scope.startswith("group_") else "private",
group_id=int(scope.split("_", 1)[1]) if scope.startswith("group_") else None, group_id=int(scope.split("_", 1)[1]) if scope.startswith("group_") else None,
user_id=int(scope.split("_", 1)[1]) if scope.startswith("test_") else None, user_id=int(scope.split("_", 1)[1]) if scope.startswith("test_") else None,
message=message, message=outbound_message,
) )
except Exception: except Exception:
pass pass

View File

@@ -22,6 +22,11 @@ class Config(BaseSettings):
MIN_ODDS: float = 1.2 MIN_ODDS: float = 1.2
RACE_DISTANCE: int = 100 RACE_DISTANCE: int = 100
RACE_TICK_INTERVAL: int = 5 RACE_TICK_INTERVAL: int = 5
RACE_RENDER_AS_IMAGE: bool = True
RACE_IMAGE_WIDTH: int = 900
RACE_IMAGE_FONT_SIZE: int = 26
RACE_IMAGE_PADDING: int = 28
RACE_IMAGE_LINE_SPACING: float = 1.35
# 消息撤回配置 # 消息撤回配置
MESSAGE_RECALL: dict[str, int] = Field( MESSAGE_RECALL: dict[str, int] = Field(

View File

@@ -250,7 +250,7 @@ async def handle_test_simulate_race(bot: Bot, event: Event):
progress_count += 1 progress_count += 1
if progress_count > max_progress: if progress_count > max_progress:
return return
await test_simulate_race_cmd.send(message) await original_send_to_scope(bot, scope, message)
commands_mod._send_to_scope = _test_send_to_scope commands_mod._send_to_scope = _test_send_to_scope